The grant option check for path privileges was not correctly logged in the audit log

Author: shuwenwei

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/security/TreeAccessCheckVisitor.java

@@ -167,6 +167,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.StringJoiner;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
@@ -668,31 +669,12 @@ public TSStatus visitAuthor(AuthorStatement statement, TreeAccessCheckContext co
             auditObject)) {
           return RpcUtils.SUCCESS_STATUS;
         }
-        for (String s : statement.getPrivilegeList()) {
-          PrivilegeType privilegeType = PrivilegeType.valueOf(s.toUpperCase());
-          if (privilegeType.isSystemPrivilege()) {
-            if (!checkHasGlobalAuth(context, privilegeType, auditObject, true)) {
-              return AuthorityChecker.getTSStatus(
-                  false,
-                  "Has no permission to execute "
-                      + authorType
-                      + ", please ensure you have these privileges and the grant option is TRUE when granted)");
-            }
-          } else if (privilegeType.isPathPrivilege()) {
-            if (!AuthorityChecker.checkPathPermissionGrantOption(
-                context.getUsername(), privilegeType, statement.getNodeNameList())) {
-              return AuthorityChecker.getTSStatus(
-                  false,
-                  "Has no permission to execute "
-                      + authorType
-                      + ", please ensure you have these privileges and the grant option is TRUE when granted)");
-            }
-          } else {
-            return AuthorityChecker.getTSStatus(
-                false, "Not support Relation statement in tree sql_dialect");
-          }
-        }
-        return RpcUtils.SUCCESS_STATUS;
+        return checkPermissionsWithGrantOption(
+            context,
+            Arrays.stream(statement.getPrivilegeList())
+                .map(PrivilegeType::valueOf)
+                .collect(Collectors.toList()),
+            statement.getNodeNameList());
       default:
         throw new IllegalArgumentException("Unknown authorType: " + authorType);
     }
@@ -1997,6 +1979,58 @@ protected boolean checkHasGlobalAuth(
     return result;
   }
 
+  protected TSStatus checkPermissionsWithGrantOption(
+      IAuditEntity auditEntity, List<PrivilegeType> privilegeList, List<PartialPath> paths) {
+    Supplier<String> supplier =
+        () -> {
+          StringJoiner joiner = new StringJoiner(" ");
+          if (paths != null) {
+            paths.forEach(path -> joiner.add(path.getFullPath()));
+          }
+          return joiner.toString();
+        };
+    auditEntity.setPrivilegeTypes(privilegeList);
+    if (AuthorityChecker.SUPER_USER.equals(auditEntity.getUsername())) {
+      recordObjectAuthenticationAuditLog(auditEntity.setResult(true), supplier);
+      return SUCCEED;
+    }
+    TSStatus status = SUCCEED;
+    for (PrivilegeType privilegeType : privilegeList) {
+      if (privilegeType.isSystemPrivilege()) {
+        if (!AuthorityChecker.checkSystemPermissionGrantOption(
+            auditEntity.getUsername(), privilegeType)) {
+          status =
+              AuthorityChecker.getTSStatus(
+                  false,
+                  "Has no permission to execute "
+                      + privilegeType
+                      + ", please ensure you have these privileges and the grant option is TRUE when granted)");
+          break;
+        }
+      } else if (privilegeType.isPathPrivilege()) {
+        if (!AuthorityChecker.checkPathPermissionGrantOption(
+            auditEntity.getUsername(), privilegeType, paths)) {
+          status =
+              AuthorityChecker.getTSStatus(
+                  false,
+                  "Has no permission to execute "
+                      + privilegeType
+                      + ", please ensure you have these privileges and the grant option is TRUE when granted)");
+          break;
+        }
+      } else {
+        status =
+            AuthorityChecker.getTSStatus(
+                false, "Not support Relation statement in tree sql_dialect");
+        break;
+      }
+    }
+    recordObjectAuthenticationAuditLog(
+        auditEntity.setResult(status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()),
+        supplier);
+    return status;
+  }
+
   protected TSStatus checkWriteOnReadOnlyPath(IAuditEntity auditEntity, PartialPath path) {
     if (includeByAuditTreeDB(path)
         && !AuthorityChecker.INTERNAL_AUDIT_USER.equals(path.getFullPath())) {