[enhancement](ldap) optimize LDAP authentication. (#11948)

* [enhancement](ldap) optimize LDAP authentication. 1. Support caching LDAP user information. 2. HTTP authentication supports LDAP. 3. LDAP temporary users support default user property. 4. LDAP configuration supports the `admin show config` and `admin set config` commands.
2022-08-24 17:08:14 +08:00
parent d87ab69ead
commit b619bb2000
21 changed files with 593 additions and 242 deletions
--- a/build.sh
+++ b/build.sh
@ -432,6 +432,7 @@ if [[ "${BUILD_FE}" -eq 1 ]]; then

    cp -r -p "${DORIS_HOME}/bin"/*_fe.sh "${DORIS_OUTPUT}/fe/bin"/
    cp -r -p "${DORIS_HOME}/conf/fe.conf" "${DORIS_OUTPUT}/fe/conf"/
+    cp -r -p "${DORIS_HOME}/conf/ldap.conf" "${DORIS_OUTPUT}/fe/conf"/
    cp -r -p "${DORIS_HOME}/conf"/*.xml "${DORIS_OUTPUT}/fe/conf"/
    rm -rf "${DORIS_OUTPUT}/fe/lib"/*
    cp -r -p "${DORIS_HOME}/fe/fe-core/target/lib"/* "${DORIS_OUTPUT}/fe/lib"/
--- a/fe/conf/ldap.conf
+++ b/fe/conf/ldap.conf
@ -41,14 +41,16 @@ ldap_user_basedn = ou=people,dc=domain,dc=com
 ldap_user_filter = (&(uid={login}))
 ldap_group_basedn = ou=group,dc=domain,dc=com

+# ldap_cache_time_out_s = 12 * 60 * 60;
+
 # LDAP pool configuration
 # https://docs.spring.io/spring-ldap/docs/2.3.3.RELEASE/reference/#pool-configuration
-#max_active = 8
-#max_total = -1
-#max_idle = 8
-#min_idle = 0
-#max_wait = -1
-#when_exhausted = 1
-#test_on_borrow = false
-#test_on_return = false
-#test_while_idle = false
+# ldap_pool_max_active = 8
+# ldap_pool_max_total = -1
+# ldap_pool_max_idle = 8
+# ldap_pool_min_idle = 0
+# ldap_pool_max_wait = -1
+# ldap_pool_when_exhausted = 1
+# ldap_pool_test_on_borrow = false
+# ldap_pool_test_on_return = false
+# ldap_pool_test_while_idle = false
--- a/docs/en/docs/admin-manual/privilege-ldap/ldap.md
+++ b/docs/en/docs/admin-manual/privilege-ldap/ldap.md
@ -171,5 +171,3 @@ If jack also belongs to the LDAP groups doris_qa, doris_pm; Doris exists roles:
 ## Limitations of LDAP authentication

 * The current LDAP feature of Doris only supports plaintext password authentication, that is, when a user logs in, the password is transmitted in plaintext between client and fe and between fe and LDAP service.
-* The current LDAP authentication only supports password authentication under mysql protocol. If you use the Http interface, you cannot use LDAP users for authentication.
-* Temporary users do not have user properties.
--- a/docs/zh-CN/docs/admin-manual/privilege-ldap/ldap.md
+++ b/docs/zh-CN/docs/admin-manual/privilege-ldap/ldap.md
@ -188,5 +188,3 @@ member: uid=jack,ou=aidp,dc=domain,dc=com
 ## LDAP验证的局限

 - 目前Doris的LDAP功能只支持明文密码验证，即用户登录时，密码在client与fe之间、fe与LDAP服务之间以明文的形式传输。
- 当前的LDAP验证只支持在mysql协议下进行密码验证，如果使用Http接口则无法使用LDAP用户进行验证。
- 临时用户不具有用户属性。
--- a/fe/fe-core/src/main/java/org/apache/doris/common/ConfigBase.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/ConfigBase.java
@ -75,11 +75,12 @@ public class ConfigBase {
    private static String ldapCustomConfFile;
    public static Class<? extends ConfigBase> ldapConfClass;

+    public static Map<String, Field> ldapConfFields = Maps.newHashMap();
+
    private boolean isLdapConfig = false;

    public void init(String configFile) throws Exception {
-        this.isLdapConfig = (this instanceof LdapConfig);
-        if (!isLdapConfig) {
+        if (this instanceof Config) {
            confClass = this.getClass();
            confFile = configFile;
            confFields = Maps.newHashMap();
@ -92,9 +93,17 @@ public class ConfigBase {
            }

            initConf(confFile);
-        } else {
+        } else if (this instanceof LdapConfig) {
+            isLdapConfig = true;
            ldapConfClass = this.getClass();
            ldapConfFile = configFile;
+            for (Field field : ldapConfClass.getFields()) {
+                ConfField confField = field.getAnnotation(ConfField.class);
+                if (confField == null) {
+                    continue;
+                }
+                ldapConfFields.put(confField.value().equals("") ? field.getName() : confField.value(), field);
+            }
            initConf(ldapConfFile);
        }
    }
@ -292,7 +301,11 @@ public class ConfigBase {
    public static synchronized void setMutableConfig(String key, String value) throws DdlException {
        Field field = confFields.get(key);
        if (field == null) {
-            throw new DdlException("Config '" + key + "' does not exist");
+            if (ldapConfFields.containsKey(key)) {
+                field = ldapConfFields.get(key);
+            } else {
+                throw new DdlException("Config '" + key + "' does not exist");
+            }
        }

        ConfField anno = field.getAnnotation(ConfField.class);
@ -313,7 +326,10 @@ public class ConfigBase {
    }

    public static synchronized List<List<String>> getConfigInfo(PatternMatcher matcher) {
-        return confFields.entrySet().stream().sorted(Map.Entry.comparingByKey()).flatMap(e -> {
+        Map<String, Field> allConfFields = Maps.newHashMap();
+        allConfFields.putAll(confFields);
+        allConfFields.putAll(ldapConfFields);
+        return allConfFields.entrySet().stream().sorted(Map.Entry.comparingByKey()).flatMap(e -> {
            String confKey = e.getKey();
            Field f = e.getValue();
            ConfField anno = f.getAnnotation(ConfField.class);
--- a/fe/fe-core/src/main/java/org/apache/doris/common/LdapConfig.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/LdapConfig.java
@ -66,10 +66,18 @@ public class LdapConfig extends ConfigBase {
    public static String ldap_group_basedn = "";

    /**
-     * Maximum number of user connections. This value should be between 1 and 10000.
+     * The user LDAP information cache time.
+     * After timeout, the user information will be retrieved from the LDAP service again.
     */
-    @ConfigBase.ConfField
-    public static long user_max_connections = 100L;
+    @ConfigBase.ConfField(mutable = true)
+    public static long ldap_user_cache_timeout_s = 12 * 60 * 60;
+
+    /**
+     * System LDAP information cache time.
+     * After timeout, clear all user information in the cache.
+     */
+    @ConfigBase.ConfField(mutable = true)
+    public static long ldap_cache_timeout_day = 30;

    /**
     * LDAP pool configuration:
@ -80,35 +88,35 @@ public class LdapConfig extends ConfigBase {
     * from this pool at the same time. You can use a non-positive number for no limit.
     */
    @ConfigBase.ConfField
-    public static int max_active = 8;
+    public static int ldap_pool_max_active = 8;

    /**
     * The overall maximum number of active connections (for all types) that can be allocated from this pool
     * at the same time. You can use a non-positive number for no limit.
     */
    @ConfigBase.ConfField
-    public static int max_total = -1;
+    public static int ldap_pool_max_total = -1;

    /**
     * The maximum number of active connections of each type (read-only or read-write) that can remain idle
     * in the pool without extra connections being released. You can use a non-positive number for no limit.
     */
    @ConfigBase.ConfField
-    public static int max_idle = 8;
+    public static int ldap_pool_max_idle = 8;

    /**
     * The minimum number of active connections of each type (read-only or read-write) that can remain idle
     * in the pool without extra connections being created. You can use zero (the default) to create none.
     */
    @ConfigBase.ConfField
-    public static int min_idle = 0;
+    public static int ldap_pool_min_idle = 0;

    /**
     * The maximum number of milliseconds that the pool waits (when no connections are available) for a connection
     * to be returned before throwing an exception. You can use a non-positive number to wait indefinitely.
     */
    @ConfigBase.ConfField
-    public static int max_wait = -1;
+    public static int ldap_pool_max_wait = -1;

    /**
     * Specifies the behavior when the pool is exhausted.
@ -121,25 +129,25 @@ public class LdapConfig extends ConfigBase {
     * The '2' option creates and returns a new object (essentially making max-active meaningless).
     */
    @ConfigBase.ConfField
-    public static byte when_exhausted = 1;
+    public static byte ldap_pool_when_exhausted = 1;

    /**
     * Whether objects are validated before being borrowed from the pool. If the object fails to validate,
     * it is dropped from the pool, and an attempt to borrow another is made.
     */
    @ConfigBase.ConfField
-    public static boolean test_on_borrow = false;
+    public static boolean ldap_pool_test_on_borrow = false;

    /**
     * Whether objects are validated before being returned to the pool.
     */
    @ConfigBase.ConfField
-    public static boolean test_on_return = false;
+    public static boolean ldap_pool_test_on_return = false;

    /**
     * Whether objects are validated by the idle object evictor (if any). If an object fails to validate,
     * it is dropped from the pool.
     */
    @ConfigBase.ConfField
-    public static boolean test_while_idle = false;
+    public static boolean ldap_pool_test_while_idle = false;
 }
--- a/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapAuthenticate.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapAuthenticate.java
@ -22,17 +22,12 @@ import org.apache.doris.catalog.Env;
 import org.apache.doris.cluster.ClusterNamespace;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
-import org.apache.doris.common.LdapConfig;
-import org.apache.doris.mysql.privilege.PaloRole;
 import org.apache.doris.qe.ConnectContext;

 import com.google.common.base.Strings;
-import com.google.common.collect.Lists;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;

-import java.util.List;
-
 /**
 * This class is used for LDAP authentication login and LDAP group authorization.
 * This means that users can log in to Doris with a user name and LDAP password,
@ -41,20 +36,6 @@ import java.util.List;
 public class LdapAuthenticate {
    private static final Logger LOG = LogManager.getLogger(LdapAuthenticate.class);

-    private static final String LDAP_GROUPS_PRIVS_NAME = "ldapGroupsPrivs";
-
-    // Maximum number of the user LDAP authentication login connections.
-    private static long userMaxConn = 100;
-
-    {
-        if (LdapConfig.user_max_connections <= 0 || LdapConfig.user_max_connections > 10000) {
-            LOG.warn("Ldap config user_max_connections is invalid. It should be set between 1 and 10000. "
-                    + "And now, it is set to the default value.");
-        } else {
-            userMaxConn = LdapConfig.user_max_connections;
-        }
-    }
-
    /**
     * The LDAP authentication process is as follows:
     * step1: Check the LDAP password.
@ -70,8 +51,8 @@ public class LdapAuthenticate {

        // check user password by ldap server.
        try {
-            if (!LdapClient.checkPassword(userName, password)) {
-                LOG.debug("user:{} use error LDAP password.", userName);
+            if (!Env.getCurrentEnv().getAuth().getLdapManager().checkUserPasswd(qualifiedUser, password)) {
+                LOG.info("user:{} use check LDAP password failed.", userName);
                ErrorReport.report(ErrorCode.ERR_ACCESS_DENIED_ERROR, qualifiedUser, context.getRemoteIP(), usePasswd);
                return false;
            }
@ -80,15 +61,6 @@ public class LdapAuthenticate {
            return false;
        }

-        // Get the LDAP groups privileges as a role.
-        PaloRole ldapGroupsPrivs;
-        try {
-            ldapGroupsPrivs = getLdapGroupsPrivs(userName, clusterName);
-        } catch (Exception e) {
-            LOG.error("Get ldap groups error.", e);
-            return false;
-        }
-
        String remoteIp = context.getMysqlChannel().getRemoteIp();
        UserIdentity tempUserIdentity = UserIdentity.createAnalyzedUserIdentWithIp(qualifiedUser, remoteIp);
        // Search the user in doris.
@ -97,48 +69,11 @@ public class LdapAuthenticate {
            userIdentity = tempUserIdentity;
            LOG.debug("User:{} does not exists in doris, login as temporary users.", userName);
            context.setIsTempUser(true);
-            if (ldapGroupsPrivs == null) {
-                ldapGroupsPrivs = new PaloRole(LDAP_GROUPS_PRIVS_NAME);
-            }
-            LdapPrivsChecker.grantDefaultPrivToTempUser(ldapGroupsPrivs, clusterName);
        }

        context.setCurrentUserIdentity(userIdentity);
        context.setRemoteIP(remoteIp);
-        context.setLdapGroupsPrivs(ldapGroupsPrivs);
-        LOG.debug("ldap authentication success: identity:{}, privs:{}",
-                context.getCurrentUserIdentity(), context.getLdapGroupsPrivs());
+        LOG.debug("ldap authentication success: identity:{}", context.getCurrentUserIdentity());
        return true;
    }
-
-    /**
-     * Step1: get ldap groups from ldap server;
-     * Step2: get roles by ldap groups;
-     * Step3: merge the roles;
-     */
-    private static PaloRole getLdapGroupsPrivs(String userName, String clusterName) {
-        //get user ldap group. the ldap group name should be the same as the doris role name
-        List<String> ldapGroups = LdapClient.getGroups(userName);
-        List<String> rolesNames = Lists.newArrayList();
-        for (String group : ldapGroups) {
-            String qualifiedRole = ClusterNamespace.getFullName(clusterName, group);
-            if (Env.getCurrentEnv().getAuth().doesRoleExist(qualifiedRole)) {
-                rolesNames.add(qualifiedRole);
-            }
-        }
-        LOG.debug("get user:{} ldap groups:{} and doris roles:{}", userName, ldapGroups, rolesNames);
-
-        // merge the roles
-        if (rolesNames.isEmpty()) {
-            return null;
-        } else {
-            PaloRole ldapGroupsPrivs = new PaloRole(LDAP_GROUPS_PRIVS_NAME);
-            Env.getCurrentEnv().getAuth().mergeRolesNoCheckName(rolesNames, ldapGroupsPrivs);
-            return ldapGroupsPrivs;
-        }
-    }
-
-    public static long getMaxConn() {
-        return userMaxConn;
-    }
 }
--- a/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapClient.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapClient.java
@ -43,7 +43,7 @@ import java.util.List;
 public class LdapClient {
    private static final Logger LOG = LogManager.getLogger(LdapClient.class);

-    private static volatile ClientInfo clientInfo;
+    private volatile ClientInfo clientInfo;

    @Data
    private static class ClientInfo {
@ -85,15 +85,15 @@ public class LdapClient {
            PoolingContextSource poolingContextSource = new PoolingContextSource();
            poolingContextSource.setDirContextValidator(new DefaultDirContextValidator());
            poolingContextSource.setContextSource(contextSource);
-            poolingContextSource.setMaxActive(LdapConfig.max_active);
-            poolingContextSource.setMaxTotal(LdapConfig.max_total);
-            poolingContextSource.setMaxIdle(LdapConfig.max_idle);
-            poolingContextSource.setMaxWait(LdapConfig.max_wait);
-            poolingContextSource.setMinIdle(LdapConfig.min_idle);
-            poolingContextSource.setWhenExhaustedAction(LdapConfig.when_exhausted);
-            poolingContextSource.setTestOnBorrow(LdapConfig.test_on_borrow);
-            poolingContextSource.setTestOnReturn(LdapConfig.test_on_return);
-            poolingContextSource.setTestWhileIdle(LdapConfig.test_while_idle);
+            poolingContextSource.setMaxActive(LdapConfig.ldap_pool_max_active);
+            poolingContextSource.setMaxTotal(LdapConfig.ldap_pool_max_total);
+            poolingContextSource.setMaxIdle(LdapConfig.ldap_pool_max_idle);
+            poolingContextSource.setMaxWait(LdapConfig.ldap_pool_max_wait);
+            poolingContextSource.setMinIdle(LdapConfig.ldap_pool_min_idle);
+            poolingContextSource.setWhenExhaustedAction(LdapConfig.ldap_pool_when_exhausted);
+            poolingContextSource.setTestOnBorrow(LdapConfig.ldap_pool_test_on_borrow);
+            poolingContextSource.setTestOnReturn(LdapConfig.ldap_pool_test_on_return);
+            poolingContextSource.setTestWhileIdle(LdapConfig.ldap_pool_test_while_idle);

            TransactionAwareContextSourceProxy proxy = new TransactionAwareContextSourceProxy(poolingContextSource);
            ldapTemplatePool = new LdapTemplate(proxy);
@ -104,7 +104,7 @@ public class LdapClient {
        }
    }

-    private static void init() {
+    private void init() {
        LdapInfo ldapInfo = Env.getCurrentEnv().getAuth().getLdapInfo();
        if (ldapInfo == null || !ldapInfo.isValid()) {
            LOG.error("info is null, maybe no ldap admin password is set.");
@ -123,7 +123,7 @@ public class LdapClient {
        }
    }

-    public static boolean doesUserExist(String userName) {
+    boolean doesUserExist(String userName) {
        String user = getUserDn(userName);
        if (user == null) {
            LOG.debug("User:{} does not exist in LDAP.", userName);
@ -132,7 +132,7 @@ public class LdapClient {
        return true;
    }

-    public static boolean checkPassword(String userName, String password) {
+    boolean checkPassword(String userName, String password) {
        init();
        try {
            clientInfo.getLdapTemplateNoPool().authenticate(org.springframework.ldap.query.LdapQueryBuilder.query()
@ -145,7 +145,7 @@ public class LdapClient {
    }

    // Search group DNs by 'member' attribution.
-    public static List<String> getGroups(String userName) {
+    List<String> getGroups(String userName) {
        List<String> groups = Lists.newArrayList();
        if (LdapConfig.ldap_group_basedn.isEmpty()) {
            return groups;
@ -171,7 +171,7 @@ public class LdapClient {
        return groups;
    }

-    private static String getUserDn(String userName) {
+    private String getUserDn(String userName) {
        List<String> userDns = getDn(org.springframework.ldap.query.LdapQueryBuilder.query()
                .base(LdapConfig.ldap_user_basedn).filter(getUserFilter(LdapConfig.ldap_user_filter, userName)));
        if (userDns == null || userDns.isEmpty()) {
@ -186,7 +186,7 @@ public class LdapClient {
        return userDns.get(0);
    }

-    private static List<String> getDn(LdapQuery query) {
+    private List<String> getDn(LdapQuery query) {
        init();
        try {
            return clientInfo.getLdapTemplatePool().search(query, new AbstractContextMapper<String>() {
@ -201,7 +201,7 @@ public class LdapClient {
        }
    }

-    private static String getUserFilter(String userFilter, String userName) {
+    private String getUserFilter(String userFilter, String userName) {
        return userFilter.replaceAll("\\{login}", userName);
    }
 }
--- a/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapManager.java
@ -0,0 +1,216 @@
+// 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.ldap;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.cluster.ClusterNamespace;
+import org.apache.doris.common.LdapConfig;
+import org.apache.doris.mysql.privilege.PaloAuth;
+import org.apache.doris.mysql.privilege.PaloRole;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.parquet.Strings;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * Encapsulates LDAP service interfaces and caches user LDAP information.
+ */
+public class LdapManager {
+    private static final Logger LOG = LogManager.getLogger(LdapManager.class);
+
+    private static final String LDAP_GROUPS_PRIVS_NAME = "ldapGroupsPrivs";
+
+    private final LdapClient ldapClient = new LdapClient();
+
+    private final Map<String, LdapUserInfo> ldapUserInfoCache = Maps.newHashMap();
+
+    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+    private void readLock() {
+        lock.readLock().lock();
+    }
+
+    private void readUnlock() {
+        lock.readLock().unlock();
+    }
+
+    private void writeLock() {
+        lock.writeLock().lock();
+    }
+
+    private void writeUnlock() {
+        lock.writeLock().unlock();
+    }
+
+    private volatile long lastTimestamp = System.currentTimeMillis();
+
+    // If the user exists in LDAP, the LDAP information of the user is returned; otherwise, null is returned.
+    public LdapUserInfo getUserInfo(String fullName) {
+        if (!checkParam(fullName)) {
+            return null;
+        }
+        LdapUserInfo ldapUserInfo = getUserInfoFromCache(fullName);
+        if (ldapUserInfo != null && !ldapUserInfo.checkTimeout()) {
+            return ldapUserInfo;
+        }
+        return getUserInfoAndUpdateCache(fullName);
+    }
+
+    public boolean doesUserExist(String fullName) {
+        if (!checkParam(fullName)) {
+            return false;
+        }
+        return !Objects.isNull(getUserInfo(fullName));
+    }
+
+    public boolean checkUserPasswd(String fullName, String passwd) {
+        String userName = ClusterNamespace.getNameFromFullName(fullName);
+        if (!LdapConfig.ldap_authentication_enabled || Strings.isNullOrEmpty(userName) || Objects.isNull(passwd)) {
+            return false;
+        }
+        LdapUserInfo ldapUserInfo = getUserInfo(fullName);
+        if (Objects.isNull(ldapUserInfo)) {
+            return false;
+        }
+
+        if (ldapUserInfo.isSetPasswd() && ldapUserInfo.getPasswd().equals(passwd)) {
+            return true;
+        }
+        boolean isRightPasswd = ldapClient.checkPassword(userName, passwd);
+        if (!isRightPasswd) {
+            return false;
+        }
+        updatePasswd(ldapUserInfo, passwd);
+        return true;
+    }
+
+    public boolean checkUserPasswd(String fullName, String passwd, String remoteIp, List<UserIdentity> currentUser) {
+        if (checkUserPasswd(fullName, passwd)) {
+            currentUser.add(UserIdentity.createAnalyzedUserIdentWithIp(fullName, remoteIp));
+            return true;
+        }
+        return false;
+    }
+
+    private boolean checkParam(String fullName) {
+        return LdapConfig.ldap_authentication_enabled && !Strings.isNullOrEmpty(fullName) && !fullName.equalsIgnoreCase(
+                PaloAuth.ROOT_USER) && !fullName.equalsIgnoreCase(PaloAuth.ADMIN_USER);
+    }
+
+    private LdapUserInfo getUserInfoAndUpdateCache(String fulName) {
+        String cluster = ClusterNamespace.getClusterNameFromFullName(fulName);
+        String userName = ClusterNamespace.getNameFromFullName(fulName);
+        if (Strings.isNullOrEmpty(userName)) {
+            return null;
+        } else if (!ldapClient.doesUserExist(userName)) {
+            removeUserIfExist(fulName);
+            return null;
+        }
+        checkTimeoutCleanCache();
+
+        LdapUserInfo ldapUserInfo = new LdapUserInfo(fulName, false, "", getLdapGroupsPrivs(userName, cluster));
+        writeLock();
+        try {
+            ldapUserInfoCache.put(ldapUserInfo.getUserName(), ldapUserInfo);
+        } finally {
+            writeUnlock();
+        }
+        return ldapUserInfo;
+    }
+
+    private void updatePasswd(LdapUserInfo ldapUserInfo, String passwd) {
+        LdapUserInfo newLdapUserInfo = ldapUserInfo.cloneWithPasswd(passwd);
+        writeLock();
+        try {
+            ldapUserInfoCache.put(newLdapUserInfo.getUserName(), newLdapUserInfo);
+        } finally {
+            writeUnlock();
+        }
+    }
+
+    private void removeUserIfExist(String fullName) {
+        LdapUserInfo ldapUserInfo = getUserInfoFromCache(fullName);
+        if (ldapUserInfo == null) {
+            return;
+        }
+
+        writeLock();
+        try {
+            ldapUserInfoCache.remove(ldapUserInfo.getUserName());
+        } finally {
+            writeUnlock();
+        }
+    }
+
+    private void checkTimeoutCleanCache() {
+        long tempTimestamp = System.currentTimeMillis() - LdapConfig.ldap_cache_timeout_day * 24 * 60 * 60 * 1000;
+        if (lastTimestamp < tempTimestamp) {
+            writeLock();
+            try {
+                if (lastTimestamp < tempTimestamp) {
+                    ldapUserInfoCache.clear();
+                    lastTimestamp = System.currentTimeMillis();
+                }
+            } finally {
+                writeUnlock();
+            }
+        }
+    }
+
+    private LdapUserInfo getUserInfoFromCache(String fullName) {
+        readLock();
+        try {
+            return ldapUserInfoCache.get(fullName);
+        } finally {
+            readUnlock();
+        }
+    }
+
+    /**
+     * Step1: get ldap groups from ldap server;
+     * Step2: get roles by ldap groups;
+     * Step3: merge the roles;
+     */
+    private PaloRole getLdapGroupsPrivs(String userName, String clusterName) {
+        //get user ldap group. the ldap group name should be the same as the doris role name
+        List<String> ldapGroups = ldapClient.getGroups(userName);
+        List<String> rolesNames = Lists.newArrayList();
+        for (String group : ldapGroups) {
+            String qualifiedRole = ClusterNamespace.getFullName(clusterName, group);
+            if (Env.getCurrentEnv().getAuth().doesRoleExist(qualifiedRole)) {
+                rolesNames.add(qualifiedRole);
+            }
+        }
+        LOG.debug("get user:{} ldap groups:{} and doris roles:{}", userName, ldapGroups, rolesNames);
+
+        PaloRole ldapGroupsPrivs = new PaloRole(LDAP_GROUPS_PRIVS_NAME);
+        LdapPrivsChecker.grantDefaultPrivToTempUser(ldapGroupsPrivs, clusterName);
+        if (!rolesNames.isEmpty()) {
+            Env.getCurrentEnv().getAuth().mergeRolesNoCheckName(rolesNames, ldapGroupsPrivs);
+        }
+        return ldapGroupsPrivs;
+    }
+}
--- a/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapPrivsChecker.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapPrivsChecker.java
@ -20,6 +20,7 @@ package org.apache.doris.ldap;
 import org.apache.doris.analysis.ResourcePattern;
 import org.apache.doris.analysis.TablePattern;
 import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.InfoSchemaDb;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.LdapConfig;
@ -28,7 +29,6 @@ import org.apache.doris.mysql.privilege.PaloPrivilege;
 import org.apache.doris.mysql.privilege.PaloRole;
 import org.apache.doris.mysql.privilege.PrivBitSet;
 import org.apache.doris.mysql.privilege.PrivPredicate;
-import org.apache.doris.qe.ConnectContext;

 import com.google.common.base.Preconditions;
 import com.google.common.collect.Maps;
@ -38,9 +38,7 @@ import org.apache.logging.log4j.Logger;
 import java.util.Map;

 /**
- * If the user logs in with LDAP authentication, the user LDAP group privileges will be saved in 'ldapGroupsPrivs' of
- * ConnectContext. When checking user privileges, Doris need to check both the privileges granted by Doris
- * and LDAP group privileges. This class is used for checking current user LDAP group privileges.
+ * This class is used for checking current user LDAP group privileges.
 */
 public class LdapPrivsChecker {
    private static final Logger LOG = LogManager.getLogger(LdapPrivsChecker.class);
@ -116,7 +114,7 @@ public class LdapPrivsChecker {
        if (!hasLdapPrivs(currentUser)) {
            return;
        }
-        PaloRole currentUserLdapPrivs = ConnectContext.get().getLdapGroupsPrivs();
+        PaloRole currentUserLdapPrivs = getUserLdapPrivs(currentUser.getQualifiedUser());
        for (Map.Entry<TablePattern, PrivBitSet> entry : currentUserLdapPrivs.getTblPatternToPrivs().entrySet()) {
            switch (entry.getKey().getPrivLevel()) {
                case GLOBAL:
@ -150,7 +148,7 @@ public class LdapPrivsChecker {
        if (!hasLdapPrivs(currentUser)) {
            return;
        }
-        PaloRole currentUserLdapPrivs = ConnectContext.get().getLdapGroupsPrivs();
+        PaloRole currentUserLdapPrivs = getUserLdapPrivs(currentUser.getQualifiedUser());
        for (Map.Entry<ResourcePattern, PrivBitSet> entry
                : currentUserLdapPrivs.getResourcePatternToPrivs().entrySet()) {
            switch (entry.getKey().getPrivLevel()) {
@ -177,7 +175,7 @@ public class LdapPrivsChecker {
        if (!hasLdapPrivs(currentUser)) {
            return false;
        }
-        PaloRole currentUserLdapPrivs = ConnectContext.get().getLdapGroupsPrivs();
+        PaloRole currentUserLdapPrivs = getUserLdapPrivs(currentUser.getQualifiedUser());
        for (Map.Entry<TablePattern, PrivBitSet> entry : currentUserLdapPrivs.getTblPatternToPrivs().entrySet()) {
            if (entry.getKey().getPrivLevel().equals(level) && PaloPrivilege.satisfy(entry.getValue(), wanted)) {
                return true;
@ -191,7 +189,7 @@ public class LdapPrivsChecker {
        if (!hasLdapPrivs(currentUser)) {
            return false;
        }
-        PaloRole currentUserLdapPrivs = ConnectContext.get().getLdapGroupsPrivs();
+        PaloRole currentUserLdapPrivs = getUserLdapPrivs(currentUser.getQualifiedUser());
        for (Map.Entry<TablePattern, PrivBitSet> entry : currentUserLdapPrivs.getTblPatternToPrivs().entrySet()) {
            if (entry.getKey().getPrivLevel().equals(PaloAuth.PrivLevel.TABLE)
                    && entry.getKey().getQualifiedDb().equals(db)) {
@ -201,19 +199,9 @@ public class LdapPrivsChecker {
        return false;
    }

-    public static boolean isCurrentUser(UserIdentity userIdent) {
-        ConnectContext context = ConnectContext.get();
-        if (context == null) {
-            return false;
-        }
-        UserIdentity currentUser = context.getCurrentUserIdentity();
-        return currentUser.getQualifiedUser().equals(userIdent.getQualifiedUser())
-                && currentUser.getHost().equals(userIdent.getHost());
-    }
-
    public static boolean hasLdapPrivs(UserIdentity userIdent) {
-        return LdapConfig.ldap_authentication_enabled && isCurrentUser(userIdent)
-                && ConnectContext.get().getLdapGroupsPrivs() != null;
+        return LdapConfig.ldap_authentication_enabled && Env.getCurrentEnv().getAuth().getLdapManager()
+                .doesUserExist(userIdent.getQualifiedUser());
    }

    public static Map<TablePattern, PrivBitSet> getLdapAllDbPrivs(UserIdentity userIdentity) {
@ -221,7 +209,7 @@ public class LdapPrivsChecker {
        if (!hasLdapPrivs(userIdentity)) {
            return ldapDbPrivs;
        }
-        for (Map.Entry<TablePattern, PrivBitSet> entry : ConnectContext.get().getLdapGroupsPrivs()
+        for (Map.Entry<TablePattern, PrivBitSet> entry : getUserLdapPrivs(userIdentity.getQualifiedUser())
                .getTblPatternToPrivs().entrySet()) {
            if (entry.getKey().getPrivLevel().equals(PaloAuth.PrivLevel.DATABASE)) {
                ldapDbPrivs.put(entry.getKey(), entry.getValue());
@ -235,7 +223,7 @@ public class LdapPrivsChecker {
        if (!hasLdapPrivs(userIdentity)) {
            return ldapTblPrivs;
        }
-        for (Map.Entry<TablePattern, PrivBitSet> entry : ConnectContext.get().getLdapGroupsPrivs()
+        for (Map.Entry<TablePattern, PrivBitSet> entry : getUserLdapPrivs(userIdentity.getQualifiedUser())
                .getTblPatternToPrivs().entrySet()) {
            if (entry.getKey().getPrivLevel().equals(PaloAuth.PrivLevel.TABLE)) {
                ldapTblPrivs.put(entry.getKey(), entry.getValue());
@ -249,7 +237,7 @@ public class LdapPrivsChecker {
        if (!hasLdapPrivs(userIdentity)) {
            return ldapResourcePrivs;
        }
-        for (Map.Entry<ResourcePattern, PrivBitSet> entry : ConnectContext.get().getLdapGroupsPrivs()
+        for (Map.Entry<ResourcePattern, PrivBitSet> entry : getUserLdapPrivs(userIdentity.getQualifiedUser())
                .getResourcePatternToPrivs().entrySet()) {
            if (entry.getKey().getPrivLevel().equals(PaloAuth.PrivLevel.RESOURCE)) {
                ldapResourcePrivs.put(entry.getKey(), entry.getValue());
@ -258,6 +246,10 @@ public class LdapPrivsChecker {
        return ldapResourcePrivs;
    }

+    private static PaloRole getUserLdapPrivs(String fullName) {
+        return Env.getCurrentEnv().getAuth().getLdapManager().getUserInfo(fullName).getPaloRole();
+    }
+
    // Temporary user has information_schema 'Select_priv' priv by default.
    public static void grantDefaultPrivToTempUser(PaloRole role, String clusterName) {
        TablePattern tblPattern = new TablePattern(InfoSchemaDb.DATABASE_NAME, "*");
--- a/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapUserInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapUserInfo.java
@ -0,0 +1,84 @@
+// 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.ldap;
+
+import org.apache.doris.common.LdapConfig;
+import org.apache.doris.mysql.privilege.PaloRole;
+
+import java.util.Objects;
+
+/**
+ * Used to cache LDAP information of user, such as password and privileges.
+ */
+public class LdapUserInfo {
+    public LdapUserInfo(String userName, boolean isSetPasswd, String passwd, PaloRole paloRole) {
+        this.userName = userName;
+        this.isSetPasswd = isSetPasswd;
+        this.passwd = passwd;
+        this.paloRole = paloRole;
+        this.lastTimeStamp = System.currentTimeMillis();
+    }
+
+    private LdapUserInfo(String userName, boolean isSetPasswd, String passwd, PaloRole paloRole, long lastTimeStamp) {
+        this.userName = userName;
+        this.isSetPasswd = isSetPasswd;
+        this.passwd = passwd;
+        this.paloRole = paloRole;
+        this.lastTimeStamp = lastTimeStamp;
+    }
+
+    private final String userName;
+
+    private final boolean isSetPasswd;
+
+    private final String passwd;
+
+    private final PaloRole paloRole;
+
+    private final long lastTimeStamp;
+
+    public String getUserName() {
+        return userName;
+    }
+
+    // The password needs to be checked by LdapManager for updated cache, so it is visible in the package.
+    boolean isSetPasswd() {
+        return isSetPasswd;
+    }
+
+    String getPasswd() {
+        return passwd;
+    }
+
+    public PaloRole getPaloRole() {
+        return paloRole;
+    }
+
+    public LdapUserInfo cloneWithPasswd(String passwd) {
+        if (Objects.isNull(passwd)) {
+            return new LdapUserInfo(userName, isSetPasswd, this.passwd, paloRole, lastTimeStamp);
+        }
+
+        return new LdapUserInfo(userName, true, passwd, paloRole, lastTimeStamp);
+    }
+
+    // Return true if LdapUserInfo is exceeded the time limit;
+    public boolean checkTimeout() {
+        return System.currentTimeMillis() > lastTimeStamp + LdapConfig.ldap_user_cache_timeout_s * 1000;
+    }
+}
--- a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlProto.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlProto.java
@ -26,7 +26,6 @@ import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.LdapConfig;
 import org.apache.doris.ldap.LdapAuthenticate;
-import org.apache.doris.ldap.LdapClient;
 import org.apache.doris.mysql.privilege.PaloAuth;
 import org.apache.doris.mysql.privilege.UserResource;
 import org.apache.doris.qe.ConnectContext;
@ -137,11 +136,8 @@ public class MysqlProto {
        }
        // If LDAP authentication is enabled and the user exists in LDAP, use LDAP authentication,
        // otherwise use Doris authentication.
-        if (LdapConfig.ldap_authentication_enabled
-                && LdapClient.doesUserExist(ClusterNamespace.getNameFromFullName(qualifiedUser))) {
-            return true;
-        }
-        return false;
+        return LdapConfig.ldap_authentication_enabled && Env.getCurrentEnv().getAuth().getLdapManager()
+                .doesUserExist(qualifiedUser);
    }

    /**
@ -206,7 +202,7 @@ public class MysqlProto {
        try {
            useLdapAuthenticate = useLdapAuthenticate(qualifiedUser);
        } catch (Exception e) {
-            LOG.debug("Check if user exists in ldap error.", e);
+            LOG.warn("Check if user exists in ldap error.", e);
            sendResponsePacket(context);
            return false;
        }
--- a/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/PaloAuth.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/PaloAuth.java
@ -44,6 +44,7 @@ import org.apache.doris.common.Pair;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.ldap.LdapManager;
 import org.apache.doris.ldap.LdapPrivsChecker;
 import org.apache.doris.load.DppConfig;
 import org.apache.doris.persist.LdapInfo;
@ -92,6 +93,8 @@ public class PaloAuth implements Writable {

    private LdapInfo ldapInfo = new LdapInfo();

+    private LdapManager ldapManager = new LdapManager();
+
    private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

    private void readLock() {
@ -138,6 +141,10 @@ public class PaloAuth implements Writable {
        this.ldapInfo = ldapInfo;
    }

+    public LdapManager getLdapManager() {
+        return ldapManager;
+    }
+
    private GlobalPrivEntry grantGlobalPrivs(UserIdentity userIdentity, boolean errOnExist, boolean errOnNonExist,
                                             PrivBitSet privs) throws DdlException {
        if (errOnExist && errOnNonExist) {
@ -328,6 +335,11 @@ public class PaloAuth implements Writable {
        if (!Config.enable_auth_check) {
            return true;
        }
+
+        // Check the LDAP password when the user exists in the LDAP service.
+        if (ldapManager.doesUserExist(remoteUser)) {
+            return ldapManager.checkUserPasswd(remoteUser, remotePasswd, remoteHost, currentUser);
+        }
        readLock();
        try {
            return userPrivTable.checkPlainPassword(remoteUser, remoteHost, remotePasswd, currentUser);
--- a/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/UserPropertyMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/UserPropertyMgr.java
@ -18,6 +18,7 @@
 package org.apache.doris.mysql.privilege;

 import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
@ -48,6 +49,19 @@ public class UserPropertyMgr implements Writable {
    protected Map<String, UserProperty> propertyMap = Maps.newHashMap();
    public static final String ROOT_USER = "root";
    public static final String SYSTEM_RESOURCE_USER = "system";
+    public static final String LDAP_RESOURCE_USER = "ldap";
+
+    private static final UserProperty LDAP_PROPERTY = new UserProperty(LDAP_RESOURCE_USER);
+
+    static {
+        try {
+            setNormalUserDefaultResource(LDAP_PROPERTY);
+        } catch (DdlException e) {
+            LOG.error("init DEFAULT_PROPERTY error.", e);
+            throw new RuntimeException(e);
+        }
+    }
+
    private AtomicLong resourceVersion = new AtomicLong(0);

    public UserPropertyMgr() {
@ -118,6 +132,7 @@ public class UserPropertyMgr implements Writable {

    public long getMaxConn(String qualifiedUser) {
        UserProperty existProperty = propertyMap.get(qualifiedUser);
+        existProperty = getLdapPropertyIfNull(qualifiedUser, existProperty);
        if (existProperty == null) {
            return 0;
        }
@ -126,6 +141,7 @@ public class UserPropertyMgr implements Writable {

    public long getMaxQueryInstances(String qualifiedUser) {
        UserProperty existProperty = propertyMap.get(qualifiedUser);
+        existProperty = getLdapPropertyIfNull(qualifiedUser, existProperty);
        if (existProperty == null) {
            return Config.default_max_query_instances;
        }
@ -134,6 +150,7 @@ public class UserPropertyMgr implements Writable {

    public Set<Tag> getResourceTags(String qualifiedUser) {
        UserProperty existProperty = propertyMap.get(qualifiedUser);
+        existProperty = getLdapPropertyIfNull(qualifiedUser, existProperty);
        if (existProperty == null) {
            return UserProperty.INVALID_RESOURCE_TAGS;
        }
@ -154,7 +171,7 @@ public class UserPropertyMgr implements Writable {
        userResource.updateResource("HDD_WRITE_MBPS", 30);
    }

-    private void setNormalUserDefaultResource(UserProperty user) throws DdlException {
+    private static void setNormalUserDefaultResource(UserProperty user) throws DdlException {
        UserResource userResource = user.getResource();
        userResource.updateResource("CPU_SHARE", 1000);
        userResource.updateResource("IO_SHARE", 1000);
@ -179,21 +196,21 @@ public class UserPropertyMgr implements Writable {
    public Pair<String, DppConfig> getLoadClusterInfo(String qualifiedUser, String cluster) throws DdlException {
        Pair<String, DppConfig> loadClusterInfo = null;

-        if (!propertyMap.containsKey(qualifiedUser)) {
+        UserProperty property = propertyMap.get(qualifiedUser);
+        property = getLdapPropertyIfNull(qualifiedUser, property);
+        if (property == null) {
            throw new DdlException("User " + qualifiedUser + " does not exist");
        }
-
-        UserProperty property = propertyMap.get(qualifiedUser);
        loadClusterInfo = property.getLoadClusterInfo(cluster);
        return loadClusterInfo;
    }

    public List<List<String>> fetchUserProperty(String qualifiedUser) throws AnalysisException {
-        if (!propertyMap.containsKey(qualifiedUser)) {
+        UserProperty property = propertyMap.get(qualifiedUser);
+        property = getLdapPropertyIfNull(qualifiedUser, property);
+        if (property == null) {
            throw new AnalysisException("User " + qualifiedUser + " does not exist");
        }
-
-        UserProperty property = propertyMap.get(qualifiedUser);
        return property.fetchProperty();
    }

@ -232,6 +249,7 @@ public class UserPropertyMgr implements Writable {

    public String[] getSqlBlockRules(String qualifiedUser) {
        UserProperty existProperty = propertyMap.get(qualifiedUser);
+        existProperty = getLdapPropertyIfNull(qualifiedUser, existProperty);
        if (existProperty == null) {
            return new String[]{};
        }
@ -240,18 +258,16 @@ public class UserPropertyMgr implements Writable {

    public int getCpuResourceLimit(String qualifiedUser) {
        UserProperty existProperty = propertyMap.get(qualifiedUser);
+        existProperty = getLdapPropertyIfNull(qualifiedUser, existProperty);
        if (existProperty == null) {
            return -1;
        }
        return existProperty.getCpuResourceLimit();
    }

-    public UserProperty getUserProperty(String qualifiedUserName) {
-        return propertyMap.get(qualifiedUserName);
-    }
-
    public long getExecMemLimit(String qualifiedUser) {
        UserProperty existProperty = propertyMap.get(qualifiedUser);
+        existProperty = getLdapPropertyIfNull(qualifiedUser, existProperty);
        if (existProperty == null) {
            return -1;
        }
@ -260,12 +276,20 @@ public class UserPropertyMgr implements Writable {

    public long getLoadMemLimit(String qualifiedUser) {
        UserProperty existProperty = propertyMap.get(qualifiedUser);
+        existProperty = getLdapPropertyIfNull(qualifiedUser, existProperty);
        if (existProperty == null) {
            return -1;
        }
        return existProperty.getLoadMemLimit();
    }

+    private UserProperty getLdapPropertyIfNull(String qualifiedUser, UserProperty existProperty) {
+        if (existProperty == null && Env.getCurrentEnv().getAuth().getLdapManager().doesUserExist(qualifiedUser)) {
+            return LDAP_PROPERTY;
+        }
+        return existProperty;
+    }
+
    public static UserPropertyMgr read(DataInput in) throws IOException {
        UserPropertyMgr userPropertyMgr = new UserPropertyMgr();
        userPropertyMgr.readFields(in);
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
@ -31,7 +31,6 @@ import org.apache.doris.mysql.MysqlCapability;
 import org.apache.doris.mysql.MysqlChannel;
 import org.apache.doris.mysql.MysqlCommand;
 import org.apache.doris.mysql.MysqlSerializer;
-import org.apache.doris.mysql.privilege.PaloRole;
 import org.apache.doris.plugin.AuditEvent.AuditEventBuilder;
 import org.apache.doris.resource.Tag;
 import org.apache.doris.thrift.TResourceInfo;
@ -88,8 +87,6 @@ public class ConnectContext {
    // LDAP authenticated but the Doris account does not exist,
    // set the flag, and the user login Doris as Temporary user.
    protected volatile boolean isTempUser = false;
-    // Save the privs from the ldap groups.
-    protected volatile PaloRole ldapGroupsPrivs = null;
    // username@host combination for the Doris account
    // that the server used to authenticate the current client.
    // In other word, currentUserIdentity is the entry that matched in Doris auth table.
@ -321,14 +318,6 @@ public class ConnectContext {
        this.isTempUser = isTempUser;
    }

-    public PaloRole getLdapGroupsPrivs() {
-        return ldapGroupsPrivs;
-    }
-
-    public void setLdapGroupsPrivs(PaloRole ldapGroupsPrivs) {
-        this.ldapGroupsPrivs = ldapGroupsPrivs;
-    }
-
    // for USER() function
    public UserIdentity getUserIdentity() {
        return new UserIdentity(qualifiedUser, remoteIP);
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectScheduler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectScheduler.java
@ -21,7 +21,6 @@ import org.apache.doris.catalog.Env;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ThreadPoolManager;
-import org.apache.doris.ldap.LdapAuthenticate;
 import org.apache.doris.mysql.MysqlProto;
 import org.apache.doris.mysql.nio.NConnectContext;
 import org.apache.doris.mysql.privilege.PrivPredicate;
@ -102,13 +101,7 @@ public class ConnectScheduler {
        // Check user
        connByUser.putIfAbsent(ctx.getQualifiedUser(), new AtomicInteger(0));
        AtomicInteger conns = connByUser.get(ctx.getQualifiedUser());
-        if (ctx.getIsTempUser()) {
-            if (conns.incrementAndGet() > LdapAuthenticate.getMaxConn()) {
-                conns.decrementAndGet();
-                numberConnection.decrementAndGet();
-                return false;
-            }
-        } else if (conns.incrementAndGet() > ctx.getEnv().getAuth().getMaxConn(ctx.getQualifiedUser())) {
+        if (conns.incrementAndGet() > ctx.getEnv().getAuth().getMaxConn(ctx.getQualifiedUser())) {
            conns.decrementAndGet();
            numberConnection.decrementAndGet();
            return false;
--- a/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapAuthenticateTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapAuthenticateTest.java
@ -25,7 +25,6 @@ import org.apache.doris.mysql.privilege.PaloAuth;
 import org.apache.doris.mysql.privilege.PaloRole;
 import org.apache.doris.qe.ConnectContext;

-import com.google.common.collect.Lists;
 import mockit.Delegate;
 import mockit.Expectations;
 import mockit.Mocked;
@ -33,7 +32,6 @@ import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;

-import java.util.ArrayList;
 import java.util.List;

 public class LdapAuthenticateTest {
@ -45,7 +43,7 @@ public class LdapAuthenticateTest {
    private PaloRole ldapGroupsPrivs;

    @Mocked
-    private LdapClient ldapClient;
+    private LdapManager ldapManager;
    @Mocked
    private LdapPrivsChecker ldapPrivsChecker;
    @Mocked
@ -83,7 +81,7 @@ public class LdapAuthenticateTest {
    private void setCheckPassword(boolean res) {
        new Expectations() {
            {
-                LdapClient.checkPassword(anyString, anyString);
+                ldapManager.checkUserPasswd(anyString, anyString);
                minTimes = 0;
                result = res;
            }
@ -93,45 +91,33 @@ public class LdapAuthenticateTest {
    private void setCheckPasswordException() {
        new Expectations() {
            {
-                LdapClient.checkPassword(anyString, anyString);
+                ldapManager.checkUserPasswd(anyString, anyString);
                minTimes = 0;
                result = new RuntimeException("exception");
            }
        };
    }

-    private void setGetGroups(boolean res) {
+    private void setGetUserInfo(boolean res) {
        new Expectations() {
            {
                if (res) {
-                    LdapClient.getGroups(anyString);
+                    ldapManager.getUserInfo(anyString);
                    minTimes = 0;
                    result = new Delegate() {
-                        List<String> fakeGetGroups(String user) {
-                            List<String> list = new ArrayList<>();
-                            list.add(TABLE_RD);
-                            return list;
+                        LdapUserInfo fakeGetGroups(String user) {
+                            return new LdapUserInfo(anyString, false, "", new PaloRole(anyString));
                        }
                    };
                } else {
-                    LdapClient.getGroups(anyString);
+                    ldapManager.getUserInfo(anyString);
                    minTimes = 0;
-                    result = Lists.newArrayList();
+                    result = null;
                }
            }
        };
    }

-    private void setGetGroupsException() {
-        new Expectations() {
-            {
-                LdapClient.getGroups(anyString);
-                minTimes = 0;
-                result = new RuntimeException("exception");
-            }
-        };
-    }
-
    private void setGetCurrentUserIdentity(boolean res) {
        new Expectations() {
            {
@ -160,71 +146,54 @@ public class LdapAuthenticateTest {
    public void testAuthenticate() {
        ConnectContext context = getContext();
        setCheckPassword(true);
-        setGetGroups(true);
+        setGetUserInfo(true);
        setGetCurrentUserIdentity(true);
        String qualifiedUser = ClusterNamespace.getFullName(DEFAULT_CLUSTER, USER_NAME);
        Assert.assertTrue(LdapAuthenticate.authenticate(context, "123", qualifiedUser));
        Assert.assertFalse(context.getIsTempUser());
-        Assert.assertSame(ldapGroupsPrivs, context.getLdapGroupsPrivs());
    }

    @Test
    public void testAuthenticateWithWrongPassword() {
        ConnectContext context = getContext();
        setCheckPassword(false);
-        setGetGroups(true);
+        setGetUserInfo(true);
        setGetCurrentUserIdentity(true);
        String qualifiedUser = ClusterNamespace.getFullName(DEFAULT_CLUSTER, USER_NAME);
        Assert.assertFalse(LdapAuthenticate.authenticate(context, "123", qualifiedUser));
        Assert.assertFalse(context.getIsTempUser());
-        Assert.assertNull(context.getLdapGroupsPrivs());
    }

    @Test
    public void testAuthenticateWithCheckPasswordException() {
        ConnectContext context = getContext();
        setCheckPasswordException();
-        setGetGroups(true);
+        setGetUserInfo(true);
        setGetCurrentUserIdentity(true);
        String qualifiedUser = ClusterNamespace.getFullName(DEFAULT_CLUSTER, USER_NAME);
        Assert.assertFalse(LdapAuthenticate.authenticate(context, "123", qualifiedUser));
        Assert.assertFalse(context.getIsTempUser());
-        Assert.assertNull(context.getLdapGroupsPrivs());
    }

    @Test
    public void testAuthenticateGetGroupsNull() {
        ConnectContext context = getContext();
        setCheckPassword(true);
-        setGetGroups(false);
+        setGetUserInfo(false);
        setGetCurrentUserIdentity(true);
        String qualifiedUser = ClusterNamespace.getFullName(DEFAULT_CLUSTER, USER_NAME);
        Assert.assertTrue(LdapAuthenticate.authenticate(context, "123", qualifiedUser));
        Assert.assertFalse(context.getIsTempUser());
-        Assert.assertNull(context.getLdapGroupsPrivs());
-    }
-
-    @Test
-    public void testAuthenticateGetGroupsException() {
-        ConnectContext context = getContext();
-        setCheckPassword(true);
-        setGetGroupsException();
-        setGetCurrentUserIdentity(true);
-        String qualifiedUser = ClusterNamespace.getFullName(DEFAULT_CLUSTER, USER_NAME);
-        Assert.assertFalse(LdapAuthenticate.authenticate(context, "123", qualifiedUser));
-        Assert.assertFalse(context.getIsTempUser());
-        Assert.assertNull(context.getLdapGroupsPrivs());
    }

    @Test
    public void testAuthenticateUserNotExistInDoris() {
        ConnectContext context = getContext();
        setCheckPassword(true);
-        setGetGroups(true);
+        setGetUserInfo(true);
        setGetCurrentUserIdentity(false);
        String qualifiedUser = ClusterNamespace.getFullName(DEFAULT_CLUSTER, USER_NAME);
        Assert.assertTrue(LdapAuthenticate.authenticate(context, "123", qualifiedUser));
        Assert.assertTrue(context.getIsTempUser());
-        Assert.assertSame(ldapGroupsPrivs, context.getLdapGroupsPrivs());
    }
 }
--- a/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapClientTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapClientTest.java
@ -49,6 +49,8 @@ public class LdapClientTest {

    private LdapInfo ldapInfo = new LdapInfo(ADMIN_PASSWORD);

+    private LdapClient ldapClient = new LdapClient();
+
    @Before
    public void setUp() {
        new Expectations() {
@ -96,7 +98,7 @@ public class LdapClientTest {
                        if (passwd.equals(password)) {
                            return;
                        } else {
-                            throw new RuntimeException("exception");
+                            throw new org.springframework.ldap.AuthenticationException();
                        }
                    }
                };
@ -109,13 +111,13 @@ public class LdapClientTest {
        List<String> list = Lists.newArrayList();
        list.add("zhangsan");
        mockLdapTemplateSearch(list);
-        Assert.assertTrue(LdapClient.doesUserExist("zhangsan"));
+        Assert.assertTrue(ldapClient.doesUserExist("zhangsan"));
    }

    @Test
    public void testDoesUserExistFail() {
        mockLdapTemplateSearch(null);
-        Assert.assertFalse(LdapClient.doesUserExist("zhangsan"));
+        Assert.assertFalse(ldapClient.doesUserExist("zhangsan"));
    }

    @Test(expected = RuntimeException.class)
@ -124,15 +126,15 @@ public class LdapClientTest {
        list.add("zhangsan");
        list.add("zhangsan");
        mockLdapTemplateSearch(list);
-        Assert.assertTrue(LdapClient.doesUserExist("zhangsan"));
+        Assert.assertTrue(ldapClient.doesUserExist("zhangsan"));
        Assert.fail("No Exception throws.");
    }

    @Test
    public void testCheckPassword() {
        mockLdapTemplateAuthenticate(ADMIN_PASSWORD);
-        Assert.assertTrue(LdapClient.checkPassword("zhangsan", ADMIN_PASSWORD));
-        Assert.assertFalse(LdapClient.checkPassword("zhangsan", "123"));
+        Assert.assertTrue(ldapClient.checkPassword("zhangsan", ADMIN_PASSWORD));
+        Assert.assertFalse(ldapClient.checkPassword("zhangsan", "123"));
    }

    @Test
@ -140,6 +142,6 @@ public class LdapClientTest {
        List<String> list = Lists.newArrayList();
        list.add("cn=groupName,ou=groups,dc=example,dc=com");
        mockLdapTemplateSearch(list);
-        Assert.assertEquals(1, LdapClient.getGroups("zhangsan").size());
+        Assert.assertEquals(1, ldapClient.getGroups("zhangsan").size());
    }
 }
--- a/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapManagerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapManagerTest.java
@ -0,0 +1,88 @@
+// 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.ldap;
+
+import org.apache.doris.common.LdapConfig;
+
+import mockit.Expectations;
+import mockit.Mocked;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+
+public class LdapManagerTest {
+
+    private static final String USER1 = "default_cluster:user1";
+    private static final String USER2 = "default_cluster:user2";
+
+    @Mocked
+    private LdapClient ldapClient;
+
+    @Before
+    public void setUp() {
+        LdapConfig.ldap_authentication_enabled = true;
+    }
+
+    private void mockClient(boolean userExist, boolean passwd) {
+        new Expectations() {
+            {
+                ldapClient.doesUserExist(anyString);
+                minTimes = 0;
+                result = userExist;
+
+                ldapClient.checkPassword(anyString, anyString);
+                minTimes = 0;
+                result = passwd;
+
+                ldapClient.getGroups(anyString);
+                minTimes = 0;
+                result = new ArrayList<>();
+            }
+        };
+    }
+
+    @Test
+    public void testGetUserInfo() {
+        LdapManager ldapManager = new LdapManager();
+        mockClient(true, true);
+        LdapUserInfo ldapUserInfo = ldapManager.getUserInfo(USER1);
+        Assert.assertNotNull(ldapUserInfo);
+        String paloRoleString = ldapUserInfo.getPaloRole().toString();
+        Assert.assertTrue(paloRoleString.contains("information_schema"));
+        Assert.assertTrue(paloRoleString.contains("Select_priv"));
+
+        mockClient(false, false);
+        Assert.assertNull(ldapManager.getUserInfo(USER2));
+    }
+
+    @Test
+    public void testCheckUserPasswd() {
+        LdapManager ldapManager = new LdapManager();
+        mockClient(true, true);
+        Assert.assertTrue(ldapManager.checkUserPasswd(USER1, "123"));
+        LdapUserInfo ldapUserInfo = ldapManager.getUserInfo(USER1);
+        Assert.assertNotNull(ldapUserInfo);
+        Assert.assertTrue(ldapUserInfo.isSetPasswd());
+        Assert.assertEquals("123", ldapUserInfo.getPasswd());
+
+        mockClient(true, false);
+        Assert.assertFalse(ldapManager.checkUserPasswd(USER2, "123"));
+    }
+}
--- a/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapPrivsCheckerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapPrivsCheckerTest.java
@ -20,9 +20,11 @@ package org.apache.doris.ldap;
 import org.apache.doris.analysis.ResourcePattern;
 import org.apache.doris.analysis.TablePattern;
 import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.LdapConfig;
 import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.mysql.privilege.PaloAuth;
 import org.apache.doris.mysql.privilege.PaloPrivilege;
 import org.apache.doris.mysql.privilege.PaloRole;
 import org.apache.doris.mysql.privilege.PrivBitSet;
@ -53,6 +55,15 @@ public class LdapPrivsCheckerTest {
    @Mocked
    private ConnectContext context;

+    @Mocked
+    private Env env;
+
+    @Mocked
+    private PaloAuth paloAuth;
+
+    @Mocked
+    private LdapManager ldapManager;
+
    @Before
    public void setUp() {
        LdapConfig.ldap_authentication_enabled = true;
@ -62,6 +73,18 @@ public class LdapPrivsCheckerTest {
                minTimes = 0;
                result = context;

+                Env.getCurrentEnv();
+                minTimes = 0;
+                result = env;
+
+                env.getAuth();
+                minTimes = 0;
+                result = paloAuth;
+
+                paloAuth.getLdapManager();
+                minTimes = 0;
+                result = ldapManager;
+
                PaloRole role = new PaloRole("");
                Map<TablePattern, PrivBitSet> tblPatternToPrivs = role.getTblPatternToPrivs();

@ -91,13 +114,20 @@ public class LdapPrivsCheckerTest {
                } catch (AnalysisException e) {
                    e.printStackTrace();
                }
-                context.getLdapGroupsPrivs();
+
+                UserIdentity userIdentity = UserIdentity.createAnalyzedUserIdentWithIp(USER, IP);
+
+                ldapManager.getUserInfo(userIdentity.getQualifiedUser());
                minTimes = 0;
-                result = role;
+                result = new LdapUserInfo(userIdentity.getQualifiedUser(), false, "", role);
+
+                ldapManager.doesUserExist(userIdentity.getQualifiedUser());
+                minTimes = 0;
+                result = true;

                context.getCurrentUserIdentity();
                minTimes = 0;
-                result = UserIdentity.createAnalyzedUserIdentWithIp(USER, IP);
+                result = userIdentity;
            }
        };
    }
@ -167,15 +197,6 @@ public class LdapPrivsCheckerTest {
        Assert.assertTrue(LdapPrivsChecker.hasPrivsOfDb(userIdent, CLUSTER + ":" + TABLE_DB));
    }

-    @Test
-    public void testIsCurrentUser() {
-        Assert.assertTrue(LdapPrivsChecker.isCurrentUser(userIdent));
-        Assert.assertFalse(LdapPrivsChecker.isCurrentUser(
-                UserIdentity.createAnalyzedUserIdentWithIp("default_cluster:lisi", IP)));
-        Assert.assertFalse(LdapPrivsChecker.isCurrentUser(
-                UserIdentity.createAnalyzedUserIdentWithIp(USER, "127.0.0.1")));
-    }
-
    @Test
    public void testGetLdapAllDbPrivs() throws AnalysisException {
        Map<TablePattern, PrivBitSet> allDb = LdapPrivsChecker.getLdapAllDbPrivs(userIdent);
--- a/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlProtoTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mysql/MysqlProtoTest.java
@ -25,7 +25,7 @@ import org.apache.doris.common.DdlException;
 import org.apache.doris.common.LdapConfig;
 import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.ldap.LdapAuthenticate;
-import org.apache.doris.ldap.LdapClient;
+import org.apache.doris.ldap.LdapManager;
 import org.apache.doris.mysql.privilege.PaloAuth;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.qe.ConnectContext;
@ -58,7 +58,7 @@ public class MysqlProtoTest {
    @Mocked
    private PaloAuth auth;
    @Mocked
-    private LdapClient ldapClient;
+    private LdapManager ldapManager;
    @Mocked
    private LdapAuthenticate ldapAuthenticate;
    @Mocked
@ -219,7 +219,11 @@ public class MysqlProtoTest {
                    }
                };

-                LdapClient.doesUserExist(anyString);
+                ldapManager.checkUserPasswd(anyString, anyString);
+                minTimes = 0;
+                result = userExist;
+
+                ldapManager.doesUserExist(anyString);
                minTimes = 0;
                result = userExist;
            }
@ -276,6 +280,7 @@ public class MysqlProtoTest {
        context.setEnv(env);
        context.setThreadLocalInfo();
        Assert.assertTrue(MysqlProto.negotiate(context));
+        LdapConfig.ldap_authentication_enabled = false;
    }

    @Test
@ -289,6 +294,7 @@ public class MysqlProtoTest {
        context.setEnv(env);
        context.setThreadLocalInfo();
        Assert.assertFalse(MysqlProto.negotiate(context));
+        LdapConfig.ldap_authentication_enabled = false;
    }

    @Test
@ -302,6 +308,7 @@ public class MysqlProtoTest {
        context.setEnv(env);
        context.setThreadLocalInfo();
        Assert.assertTrue(MysqlProto.negotiate(context));
+        LdapConfig.ldap_authentication_enabled = false;
    }

    @Test