[fix](kerberos)fix and refactor ugi login for kerberos and simple authentication

docker/thirdparties/docker-compose/kerberos/common/conf/doris-krb5.conf

@@ -24,7 +24,7 @@
  default_realm = LABS.TERADATA.COM
  dns_lookup_realm = false
  dns_lookup_kdc = false
- ticket_lifetime = 24h
+ ticket_lifetime = 5s
  # this setting is causing a Message stream modified (41) error when talking to KDC running on CentOS 7: https://stackoverflow.com/a/60978520
  # renew_lifetime = 7d
  forwardable = true

fe/fe-common/src/main/java/org/apache/doris/common/security/authentication/AuthenticationConfig.java

@@ -26,6 +26,7 @@ public abstract class AuthenticationConfig {
     public static String HADOOP_KERBEROS_KEYTAB = "hadoop.kerberos.keytab";
     public static String HIVE_KERBEROS_PRINCIPAL = "hive.metastore.kerberos.principal";
     public static String HIVE_KERBEROS_KEYTAB = "hive.metastore.kerberos.keytab.file";
+    public static String DORIS_KRB5_DEBUG = "doris.krb5.debug";
 
     /**
      * @return true if the config is valid, otherwise false.
@@ -57,6 +58,7 @@ public static AuthenticationConfig getKerberosConfig(Configuration conf,
             krbConfig.setKerberosPrincipal(conf.get(krbPrincipalKey));
             krbConfig.setKerberosKeytab(conf.get(krbKeytabKey));
             krbConfig.setConf(conf);
+            krbConfig.setPrintDebugLog(Boolean.parseBoolean(conf.get(DORIS_KRB5_DEBUG, "false")));
             return krbConfig;
         } else {
             // AuthType.SIMPLE

fe/fe-common/src/main/java/org/apache/doris/common/security/authentication/HadoopAuthenticator.java

@@ -0,0 +1,44 @@
+// 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.doris.common.security.authentication;
+
+import org.apache.hadoop.security.UserGroupInformation;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+
+public interface HadoopAuthenticator {
+
+    UserGroupInformation getUGI() throws IOException;
+
+    default <T> T doAs(PrivilegedExceptionAction<T> action) throws IOException {
+        try {
+            return getUGI().doAs(action);
+        } catch (InterruptedException e) {
+            throw new IOException(e);
+        }
+    }
+
+    static HadoopAuthenticator getHadoopAuthenticator(AuthenticationConfig config) {
+        if (config instanceof KerberosAuthenticationConfig) {
+            return new HadoopKerberosAuthenticator((KerberosAuthenticationConfig) config);
+        } else {
+            return new HadoopSimpleAuthenticator((SimpleAuthenticationConfig) config);
+        }
+    }
+}

fe/fe-common/src/main/java/org/apache/doris/common/security/authentication/HadoopKerberosAuthenticator.java

@@ -0,0 +1,163 @@
+// 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.doris.common.security.authentication;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import io.trino.plugin.base.authentication.KerberosTicketUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Date;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import javax.security.auth.Subject;
+import javax.security.auth.kerberos.KerberosPrincipal;
+import javax.security.auth.kerberos.KerberosTicket;
+import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.auth.login.LoginContext;
+import javax.security.auth.login.LoginException;
+
+public class HadoopKerberosAuthenticator implements HadoopAuthenticator {
+    private static final Logger LOG = LogManager.getLogger(HadoopKerberosAuthenticator.class);
+    private final KerberosAuthenticationConfig config;
+    private Subject subject;
+    private long nextRefreshTime;
+    private UserGroupInformation ugi;
+
+    public HadoopKerberosAuthenticator(KerberosAuthenticationConfig config) {
+        this.config = config;
+    }
+
+    public static void initializeAuthConfig(Configuration hadoopConf) {
+        hadoopConf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION, "true");
+        synchronized (HadoopKerberosAuthenticator.class) {
+            // avoid other catalog set conf at the same time
+            UserGroupInformation.setConfiguration(hadoopConf);
+        }
+    }
+
+    @Override
+    public synchronized UserGroupInformation getUGI() throws IOException {
+        if (ugi == null) {
+            subject = getSubject(config.getKerberosKeytab(), config.getKerberosPrincipal(), config.isPrintDebugLog());
+            ugi = Objects.requireNonNull(login(subject), "login result is null");
+            return ugi;
+        }
+        if (nextRefreshTime < System.currentTimeMillis()) {
+            long lastRefreshTime = nextRefreshTime;
+            Subject existingSubject = subject;
+            if (LOG.isDebugEnabled()) {
+                Date lastTicketEndTime = getTicketEndTime(subject);
+                LOG.debug("Current ticket expired time is {}", lastTicketEndTime);
+            }
+            // renew subject
+            Subject newSubject = getSubject(config.getKerberosKeytab(), config.getKerberosPrincipal(),
+                    config.isPrintDebugLog());
+            Objects.requireNonNull(login(newSubject), "re-login result is null");
+            // modify UGI instead of returning new UGI
+            existingSubject.getPrincipals().addAll(newSubject.getPrincipals());
+            Set<Object> privateCredentials = existingSubject.getPrivateCredentials();
+            // clear the old credentials
+            synchronized (privateCredentials) {
+                privateCredentials.clear();
+                privateCredentials.addAll(newSubject.getPrivateCredentials());
+            }
+            Set<Object> publicCredentials = existingSubject.getPublicCredentials();
+            synchronized (publicCredentials) {
+                publicCredentials.clear();
+                publicCredentials.addAll(newSubject.getPublicCredentials());
+            }
+            nextRefreshTime = calculateNextRefreshTime(newSubject);
+            if (LOG.isDebugEnabled()) {
+                Date lastTicketEndTime = getTicketEndTime(newSubject);
+                LOG.debug("Next ticket expired time is {}", lastTicketEndTime);
+                LOG.debug("Refresh kerberos ticket succeeded, last time is {}, next time is {}",
+                        lastRefreshTime, nextRefreshTime);
+            }
+        }
+        return ugi;
+    }
+
+    private UserGroupInformation login(Subject subject) throws IOException {
+        // login and get ugi when catalog is initialized
+        initializeAuthConfig(config.getConf());
+        String principal = config.getKerberosPrincipal();
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Login by kerberos authentication with principal: {}", principal);
+        }
+        return UserGroupInformation.getUGIFromSubject(subject);
+    }
+
+    private static long calculateNextRefreshTime(Subject subject) {
+        Preconditions.checkArgument(subject != null, "subject must be present in kerberos based UGI");
+        KerberosTicket tgtTicket = KerberosTicketUtils.getTicketGrantingTicket(subject);
+        return KerberosTicketUtils.getRefreshTime(tgtTicket);
+    }
+
+    private static Date getTicketEndTime(Subject subject) {
+        Preconditions.checkArgument(subject != null, "subject must be present in kerberos based UGI");
+        KerberosTicket tgtTicket = KerberosTicketUtils.getTicketGrantingTicket(subject);
+        return tgtTicket.getEndTime();
+    }
+
+    private static Subject getSubject(String keytab, String principal, boolean printDebugLog) {
+        Subject subject = new Subject(false, ImmutableSet.of(new KerberosPrincipal(principal)),
+                Collections.emptySet(), Collections.emptySet());
+        javax.security.auth.login.Configuration conf = getConfiguration(keytab, principal, printDebugLog);
+        try {
+            LoginContext loginContext = new LoginContext("", subject, null, conf);
+            loginContext.login();
+            return loginContext.getSubject();
+        } catch (LoginException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static javax.security.auth.login.Configuration getConfiguration(String keytab, String principal,
+                                                                            boolean printDebugLog) {
+        return new javax.security.auth.login.Configuration() {
+            @Override
+            public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
+                ImmutableMap.Builder<String, String> builder = ImmutableMap.<String, String>builder()
+                        .put("doNotPrompt", "true")
+                        .put("isInitiator", "true")
+                        .put("useKeyTab", "true")
+                        .put("storeKey", "true")
+                        .put("keyTab", keytab)
+                        .put("principal", principal);
+                if (printDebugLog) {
+                    builder.put("debug", "true");
+                }
+                Map<String, String> options = builder.build();
+                return new AppConfigurationEntry[]{
+                    new AppConfigurationEntry(
+                        "com.sun.security.auth.module.Krb5LoginModule",
+                        AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
+                        options)};
+            }
+        };
+    }
+}

fe/fe-common/src/main/java/org/apache/doris/common/security/authentication/HadoopSimpleAuthenticator.java

@@ -0,0 +1,47 @@
+// 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.doris.common.security.authentication;
+
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+public class HadoopSimpleAuthenticator implements HadoopAuthenticator {
+    private static final Logger LOG = LogManager.getLogger(HadoopSimpleAuthenticator.class);
+    private final UserGroupInformation ugi;
+
+    public HadoopSimpleAuthenticator(SimpleAuthenticationConfig config) {
+        String hadoopUserName = config.getUsername();
+        if (hadoopUserName == null) {
+            hadoopUserName = "hadoop";
+            config.setUsername(hadoopUserName);
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("{} is unset, use default user: hadoop", AuthenticationConfig.HADOOP_USER_NAME);
+            }
+        }
+        ugi = UserGroupInformation.createRemoteUser(hadoopUserName);
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Login by proxy user, hadoop.username: {}", hadoopUserName);
+        }
+    }
+
+    @Override
+    public UserGroupInformation getUGI() {
+        return ugi;
+    }
+}

fe/fe-common/src/main/java/org/apache/doris/common/security/authentication/HadoopUGI.java

@@ -18,15 +18,14 @@
 package org.apache.doris.common.security.authentication;
 
 import org.apache.commons.lang3.StringUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.io.IOException;
 import java.security.PrivilegedExceptionAction;
 
+@Deprecated
 public class HadoopUGI {
     private static final Logger LOG = LogManager.getLogger(HadoopUGI.class);
 
@@ -39,82 +38,30 @@ private static UserGroupInformation loginWithUGI(AuthenticationConfig config) {
         if (config == null || !config.isValid()) {
             return null;
         }
-        UserGroupInformation ugi;
         if (config instanceof KerberosAuthenticationConfig) {
-            KerberosAuthenticationConfig krbConfig = (KerberosAuthenticationConfig) config;
-            Configuration hadoopConf = krbConfig.getConf();
-            hadoopConf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION, "true");
-            hadoopConf.set(CommonConfigurationKeysPublic.HADOOP_KERBEROS_KEYTAB_LOGIN_AUTORENEWAL_ENABLED, "true");
-            UserGroupInformation.setConfiguration(hadoopConf);
-            String principal = krbConfig.getKerberosPrincipal();
             try {
-                //  login hadoop with keytab and try checking TGT
-                ugi = UserGroupInformation.getLoginUser();
-                LOG.debug("Current login user: {}", ugi.getUserName());
-                if (ugi.hasKerberosCredentials() && StringUtils.equals(ugi.getUserName(), principal)) {
-                    // if the current user is logged by kerberos and is the same user
-                    // just use checkTGTAndReloginFromKeytab because this method will only relogin
-                    // when the TGT is expired or is close to expiry
-                    ugi.checkTGTAndReloginFromKeytab();
-                    return ugi;
+                // TODO: remove after iceberg and hudi kerberos test case pass
+                try {
+                    //  login hadoop with keytab and try checking TGT
+                    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
+                    LOG.debug("Current login user: {}", ugi.getUserName());
+                    String principal = ((KerberosAuthenticationConfig) config).getKerberosPrincipal();
+                    if (ugi.hasKerberosCredentials() && StringUtils.equals(ugi.getUserName(), principal)) {
+                        // if the current user is logged by kerberos and is the same user
+                        // just use checkTGTAndReloginFromKeytab because this method will only relogin
+                        // when the TGT is expired or is close to expiry
+                        ugi.checkTGTAndReloginFromKeytab();
+                        return ugi;
+                    }
+                } catch (IOException e) {
+                    LOG.warn("A SecurityException occurs with kerberos, do login immediately.", e);
                 }
-            } catch (IOException e) {
-                LOG.warn("A SecurityException occurs with kerberos, do login immediately.", e);
-            }
-            try {
-                ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, krbConfig.getKerberosKeytab());
-                UserGroupInformation.setLoginUser(ugi);
-                LOG.debug("Login by kerberos authentication with principal: {}", principal);
-                return ugi;
+                return new HadoopKerberosAuthenticator((KerberosAuthenticationConfig) config).getUGI();
             } catch (IOException e) {
                 throw new RuntimeException(e);
             }
         } else {
-            String hadoopUserName = ((SimpleAuthenticationConfig) config).getUsername();
-            if (hadoopUserName == null) {
-                hadoopUserName = "hadoop";
-                ((SimpleAuthenticationConfig) config).setUsername(hadoopUserName);
-                LOG.debug(AuthenticationConfig.HADOOP_USER_NAME + " is unset, use default user: hadoop");
-            }
-
-            try {
-                ugi = UserGroupInformation.getLoginUser();
-                if (ugi.getUserName().equals(hadoopUserName)) {
-                    return ugi;
-                }
-            } catch (IOException e) {
-                LOG.warn("A SecurityException occurs with simple, do login immediately.", e);
-            }
-
-            ugi = UserGroupInformation.createRemoteUser(hadoopUserName);
-            UserGroupInformation.setLoginUser(ugi);
-            LOG.debug("Login by proxy user, hadoop.username: {}", hadoopUserName);
-            return ugi;
-        }
-    }
-
-    /**
-     * use for HMSExternalCatalog to login
-     * @param config auth config
-     */
-    public static void tryKrbLogin(String catalogName, AuthenticationConfig config) {
-        if (config instanceof KerberosAuthenticationConfig) {
-            KerberosAuthenticationConfig krbConfig = (KerberosAuthenticationConfig) config;
-            try {
-                Configuration hadoopConf = krbConfig.getConf();
-                hadoopConf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION, "true");
-                hadoopConf.set(CommonConfigurationKeysPublic.HADOOP_KERBEROS_KEYTAB_LOGIN_AUTORENEWAL_ENABLED, "true");
-                UserGroupInformation.setConfiguration(hadoopConf);
-                /**
-                 * Because metastore client is created by using
-                 * {@link org.apache.hadoop.hive.metastore.RetryingMetaStoreClient#getProxy}
-                 * it will relogin when TGT is expired, so we don't need to relogin manually.
-                 */
-                UserGroupInformation.loginUserFromKeytab(krbConfig.getKerberosPrincipal(),
-                        krbConfig.getKerberosKeytab());
-            } catch (IOException e) {
-                throw new RuntimeException("login with kerberos auth failed for catalog: " + catalogName, e);
-            }
+            return new HadoopSimpleAuthenticator((SimpleAuthenticationConfig) config).getUGI();
         }
     }