[Improvement](auth)(step-2) add ranger authorizer for hms catalog

docs/en/docs/lakehouse/multi-catalog/hive.md

@@ -117,6 +117,25 @@ CREATE CATALOG hive PROPERTIES (
 );
 ```
 
+<version since="dev">
+
+when connecting to Hive Metastore which is authorized by Ranger, need some properties and update FE runtime environment.
+
+1. add below properties when creating Catalog：
+
+```sql
+"access_controller.properties.ranger.service.name" = "<the ranger servive name your hms using>",
+"access_controller.class" = "org.apache.doris.catalog.authorizer.RangerHiveAccessControllerFactory",
+```
+
+2. update all FEs' runtime environment：
+   a. copy all ranger-*.xml files to <doris_home>/conf which are located in HMS/conf directory
+   b. update value of `ranger.plugin.hive.policy.cache.dir` in ranger-<ranger_service_name>-security.xml to a writable directory
+   c. add a log4j.properties to <doris_home>/conf, thus you can get logs of ranger authorizer
+   d. restart FE
+
+</version>
+
 In Doris 1.2.1 and newer, you can create a Resource that contains all these parameters, and reuse the Resource when creating new Catalogs. Here is an example:
 
 ```sql

docs/zh-CN/docs/lakehouse/multi-catalog/hive.md

@@ -113,6 +113,23 @@ CREATE CATALOG hive PROPERTIES (
 );
 ```
 
+<version since="dev">
+
+连接开启 Ranger 权限校验的 Hive Metastore 需要增加配置 & 配置环境：
+1. 创建 Catalog 时增加：
+
+```sql
+"access_controller.properties.ranger.service.name" = "<the ranger servive name your hms using>",
+"access_controller.class" = "org.apache.doris.catalog.authorizer.RangerHiveAccessControllerFactory",
+```
+2. 配置所有 FE 环境： 
+   a. 将 HMS conf 目录下的三个 Ranger 配置文件Copy到 <doris_home>/conf 目录下
+   b. 修改其中 ranger-<ranger_service_name>-security.xml 的属性 `ranger.plugin.hive.policy.cache.dir` 的值为一个可写目录
+   c. 为获取到 Ranger 鉴权本身的日志，可在 <doris_home>/conf 目录下添加配置文件 log4j.properties
+   d. 重启 FE
+
+</version>
+
 在 1.2.1 版本之后，我们也可以将这些信息通过创建一个 Resource 统一存储，然后在创建 Catalog 时使用这个 Resource。示例如下：
 
 ```sql

fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/RangerHiveAccessController.java

@@ -18,9 +18,12 @@
 package org.apache.doris.catalog.authorizer;
 
 import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.cluster.ClusterNamespace;
 import org.apache.doris.common.AuthorizationException;
+import org.apache.doris.common.ThreadPoolManager;
 import org.apache.doris.mysql.privilege.CatalogAccessController;
 import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.mysql.privilege.Role;
 
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HiveAccessControlException;
 import org.apache.logging.log4j.LogManager;
@@ -33,33 +36,51 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Date;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 public class RangerHiveAccessController implements CatalogAccessController {
     public static final String CLIENT_TYPE_DORIS = "doris";
     private static final Logger LOG = LogManager.getLogger(RangerHiveAccessController.class);
+    private static ScheduledThreadPoolExecutor logFlushTimer = ThreadPoolManager.newDaemonScheduledThreadPool(1,
+            "ranger-hive-audit-log-flusher-timer", true);
     private RangerHivePlugin hivePlugin;
     private RangerHiveAuditHandler auditHandler;
 
     public RangerHiveAccessController(Map<String, String> properties) {
         String serviceName = properties.get("ranger.service.name");
         hivePlugin = new RangerHivePlugin(serviceName);
         auditHandler = new RangerHiveAuditHandler(hivePlugin.getConfig());
+        //start a timed log flusher
+        logFlushTimer.scheduleAtFixedRate(new RangerHiveAuditLogFlusher(auditHandler), 10, 20L, TimeUnit.SECONDS);
     }
 
     private RangerAccessRequestImpl createRequest(UserIdentity currentUser, HiveAccessType accessType) {
         RangerAccessRequestImpl request = new RangerAccessRequestImpl();
-        request.setUser(currentUser.getQualifiedUser());
-        request.setUserRoles(currentUser.getRoles());
+        // currentUser.getQualifiedUser() is as of form: default_cluster:user1, only use `user1`
+        String[] userArray = currentUser.getQualifiedUser().split(":");
+        request.setUser(userArray[1]);
+        request.setClusterName(userArray[0]);
+        Set<String> roles = new HashSet<>();
+        for (String role : currentUser.getRoles()) {
+            // default role is as of form: default_role_rbac_xxx@%, not useful for Ranger
+            if (!Role.isDefaultRoleName(role)) {
+                roles.add(role);
+            }
+        }
+        request.setUserRoles(roles);
         request.setAction(accessType.name());
         if (accessType == HiveAccessType.USE) {
             request.setAccessType(RangerPolicyEngine.ANY_ACCESS);
         } else {
             request.setAccessType(accessType.name().toLowerCase());
         }
         request.setClientIPAddress(currentUser.getHost());
+        request.setClusterType(CLIENT_TYPE_DORIS);
         request.setClientType(CLIENT_TYPE_DORIS);
         request.setAccessTime(new Date());
 
@@ -68,28 +89,24 @@ private RangerAccessRequestImpl createRequest(UserIdentity currentUser, HiveAcce
 
     private void checkPrivileges(UserIdentity currentUser, HiveAccessType accessType,
                                 List<RangerHiveResource> hiveResources) throws AuthorizationException {
-        try {
-            List<RangerAccessRequest> requests = new ArrayList<>();
-            for (RangerHiveResource resource : hiveResources) {
-                RangerAccessRequestImpl request = createRequest(currentUser, accessType);
-                request.setResource(resource);
+        List<RangerAccessRequest> requests = new ArrayList<>();
+        for (RangerHiveResource resource : hiveResources) {
+            RangerAccessRequestImpl request = createRequest(currentUser, accessType);
+            request.setResource(resource);
 
-                requests.add(request);
-            }
+            requests.add(request);
+        }
 
-            Collection<RangerAccessResult> results = hivePlugin.isAccessAllowed(requests, auditHandler);
-            for (RangerAccessResult result : results) {
-                LOG.debug("match policy:" + result.getPolicyId());
-                if (!result.getIsAllowed()) {
-                    LOG.debug(result.getReason());
-                    throw new AuthorizationException(String.format(
-                        "Permission denied: user [%s] does not have privilege for [%s] command on [%s]",
-                        currentUser.getQualifiedUser(), accessType.name(),
-                        result.getAccessRequest().getResource().getAsString()));
-                }
+        Collection<RangerAccessResult> results = hivePlugin.isAccessAllowed(requests, auditHandler);
+        for (RangerAccessResult result : results) {
+            LOG.debug(String.format("request %s match policy %s", result.getAccessRequest(), result.getPolicyId()));
+            if (!result.getIsAllowed()) {
+                LOG.debug(result.getReason());
+                throw new AuthorizationException(String.format(
+                    "Permission denied: user [%s] does not have privilege for [%s] command on [%s]",
+                    result.getAccessRequest().getUser(), accessType.name(),
+                    result.getAccessRequest().getResource().getAsString()));
             }
-        } finally {
-            auditHandler.flushAudit();
         }
     }
 
@@ -99,20 +116,20 @@ private boolean checkPrivilege(UserIdentity currentUser, HiveAccessType accessTy
         request.setResource(resource);
 
         RangerAccessResult result = hivePlugin.isAccessAllowed(request, auditHandler);
-        auditHandler.flushAudit();
 
         if (result == null) {
-            LOG.warn(String.format("Error getting authorizer result, please check your ranger config. Request: %s",
-                    request));
+            LOG.warn(String.format("Error getting authorizer result, please check your ranger config. Make sure "
+                    + "ranger policy engine is initialized. Request: %s", request));
             return false;
         }
 
         if (result.getIsAllowed()) {
+            LOG.debug(String.format("request %s match policy %s", request, result.getPolicyId()));
             return true;
         } else {
             LOG.debug(String.format(
                     "Permission denied: user [%s] does not have privilege for [%s] command on [%s]",
-                    currentUser.getQualifiedUser(), accessType.name(),
+                    result.getAccessRequest().getUser(), accessType.name(),
                     result.getAccessRequest().getResource().getAsString()));
             return false;
         }
@@ -123,7 +140,6 @@ public String getFilterExpr(UserIdentity currentUser, HiveAccessType accessType,
         RangerAccessRequestImpl request = createRequest(currentUser, accessType);
         request.setResource(resource);
         RangerAccessResult result = hivePlugin.isAccessAllowed(request, auditHandler);
-        auditHandler.flushAudit();
 
         return result.getFilterExpr();
     }
@@ -133,7 +149,6 @@ public void getColumnMask(UserIdentity currentUser, HiveAccessType accessType,
         RangerAccessRequestImpl request = createRequest(currentUser, accessType);
         request.setResource(resource);
         RangerAccessResult result = hivePlugin.isAccessAllowed(request, auditHandler);
-        auditHandler.flushAudit();
 
         LOG.debug(String.format("maskType: %s, maskTypeDef: %s, maskedValue: %s", result.getMaskType(),
                 result.getMaskTypeDef(), result.getMaskedValue()));
@@ -142,7 +157,9 @@ public void getColumnMask(UserIdentity currentUser, HiveAccessType accessType,
     public HiveAccessType convertToAccessType(PrivPredicate predicate) {
         if (predicate == PrivPredicate.SHOW) {
             return HiveAccessType.USE;
-        } else if (predicate == PrivPredicate.ADMIN) {
+        } else if (predicate == PrivPredicate.SELECT) {
+            return HiveAccessType.SELECT;
+        } else if (predicate == PrivPredicate.ADMIN || predicate == PrivPredicate.ALL) {
             return HiveAccessType.ALL;
         } else if (predicate == PrivPredicate.LOAD) {
             return HiveAccessType.UPDATE;
@@ -152,27 +169,27 @@ public HiveAccessType convertToAccessType(PrivPredicate predicate) {
             return HiveAccessType.CREATE;
         } else if (predicate == PrivPredicate.DROP) {
             return HiveAccessType.DROP;
-        } else if (predicate == PrivPredicate.SELECT) {
-            return HiveAccessType.SELECT;
         } else {
             return HiveAccessType.NONE;
         }
     }
 
     @Override
     public boolean checkCtlPriv(UserIdentity currentUser, String ctl, PrivPredicate wanted) {
-        return false;
+        return true;
     }
 
     @Override
     public boolean checkDbPriv(UserIdentity currentUser, String ctl, String db, PrivPredicate wanted) {
-        RangerHiveResource resource = new RangerHiveResource(HiveObjectType.DATABASE, db);
+        RangerHiveResource resource = new RangerHiveResource(HiveObjectType.DATABASE,
+                ClusterNamespace.getNameFromFullName(db));
         return checkPrivilege(currentUser, convertToAccessType(wanted), resource);
     }
 
     @Override
     public boolean checkTblPriv(UserIdentity currentUser, String ctl, String db, String tbl, PrivPredicate wanted) {
-        RangerHiveResource resource = new RangerHiveResource(HiveObjectType.TABLE, db, tbl);
+        RangerHiveResource resource = new RangerHiveResource(HiveObjectType.TABLE,
+                ClusterNamespace.getNameFromFullName(db), tbl);
         return checkPrivilege(currentUser, convertToAccessType(wanted), resource);
     }
 
@@ -181,7 +198,8 @@ public void checkColsPriv(UserIdentity currentUser, String ctl, String db, Strin
                               PrivPredicate wanted) throws AuthorizationException {
         List<RangerHiveResource> resources = new ArrayList<>();
         for (String col : cols) {
-            RangerHiveResource resource = new RangerHiveResource(HiveObjectType.COLUMN, db, tbl, col);
+            RangerHiveResource resource = new RangerHiveResource(HiveObjectType.COLUMN,
+                    ClusterNamespace.getNameFromFullName(db), tbl, col);
             resources.add(resource);
         }
 

fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/RangerHiveAuditLogFlusher.java

@@ -0,0 +1,42 @@
+// 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.catalog.authorizer;
+
+import java.util.TimerTask;
+
+public class RangerHiveAuditLogFlusher extends TimerTask {
+
+    private RangerHiveAuditHandler auditHandler;
+
+    public RangerHiveAuditLogFlusher(RangerHiveAuditHandler auditHandler) {
+        this.auditHandler = auditHandler;
+    }
+
+    @Override
+    public void run() {
+        while (true) {
+            this.auditHandler.flushAudit();
+
+            try {
+                Thread.sleep(20000);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/Role.java

@@ -118,6 +118,10 @@ public Role(String roleName, TablePattern tablePattern, PrivBitSet tablePrivs,
 
     }
 
+    public static boolean isDefaultRoleName(String roleName) {
+        return roleName.startsWith(RoleManager.DEFAULT_ROLE_PREFIX);
+    }
+
     public String getRoleName() {
         return roleName;
     }