Pipe: Enabled config audit logger & Fixed some checks of config privilege & keep the tree model show pipes aligned with table model's & Fixed the bug that password with "-" prefix cannot be directly entered in Cli

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java

@@ -833,12 +833,8 @@ public void testPermission() {
         "803: No permissions for this operation, please add privilege SYSTEM",
         "test",
         "test123123456");
-    assertTestFail(
-        senderEnv,
-        "show pipes",
-        "803: No permissions for this operation, please add privilege SYSTEM",
-        "test",
-        "test123123456");
+    // Will not throw exception
+    executeQueryWithRetry(senderEnv, "show pipes", "test", "test123123456");
     assertNonQueryTestFail(
         senderEnv,
         "start pipe testPipe",

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java

@@ -22,7 +22,6 @@
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
 import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
-import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
 import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.db.it.utils.TestUtils;
 import org.apache.iotdb.it.env.MultiEnvFactory;
@@ -39,6 +38,7 @@
 import org.junit.runner.RunWith;
 
 import java.sql.Connection;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.Arrays;
@@ -417,10 +417,12 @@ public void testSourcePermission() {
     TestUtils.executeNonQuery(senderEnv, "revoke SYSTEM on root.** from user thulab");
 
     // A user shall only see its own pipe
-    try (final SyncConfigNodeIServiceClient client =
-        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
-      Assert.assertEquals(
-          1, client.showPipe(new TShowPipeReq().setUserName("thulab")).pipeInfoList.size());
+    try (final Connection connection = senderEnv.getConnection("thulab", "passwD@123456");
+        final Statement statement = connection.createStatement()) {
+      // Will not throw any exception
+      final ResultSet resultSet = statement.executeQuery("show pipes");
+      Assert.assertTrue(resultSet.next());
+      Assert.assertFalse(resultSet.next());
     } catch (Exception e) {
       fail(e.getMessage());
     }

iotdb-client/cli/src/main/java/org/apache/iotdb/cli/AbstractCli.java

@@ -321,9 +321,7 @@ static String[] removePasswordArgs(String[] args) {
       }
     }
     if (index >= 0) {
-      if (index + 1 >= args.length
-          || args[index + 1].startsWith("-")
-          || (keywordSet.contains(args[index + 1]))) {
+      if (index + 1 >= args.length || keywordSet.contains(args[index + 1])) {
         return ArrayUtils.remove(args, index);
       }
     }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.confignode.manager.pipe.receiver.protocol;
 
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.audit.IAuditEntity;
 import org.apache.iotdb.commons.auth.entity.PrivilegeType;
 import org.apache.iotdb.commons.auth.entity.PrivilegeUnion;
 import org.apache.iotdb.commons.conf.CommonDescriptor;
@@ -46,6 +47,7 @@
 import org.apache.iotdb.commons.schema.ttl.TTLCache;
 import org.apache.iotdb.commons.utils.PathUtils;
 import org.apache.iotdb.commons.utils.StatusUtils;
+import org.apache.iotdb.confignode.audit.CNAuditLogger;
 import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
@@ -80,6 +82,7 @@
 import org.apache.iotdb.confignode.consensus.request.write.table.view.SetViewCommentPlan;
 import org.apache.iotdb.confignode.consensus.request.write.table.view.SetViewPropertiesPlan;
 import org.apache.iotdb.confignode.consensus.request.write.template.CommitSetSchemaTemplatePlan;
+import org.apache.iotdb.confignode.consensus.request.write.template.CreateSchemaTemplatePlan;
 import org.apache.iotdb.confignode.consensus.request.write.template.ExtendSchemaTemplatePlan;
 import org.apache.iotdb.confignode.consensus.request.write.trigger.DeleteTriggerInTablePlan;
 import org.apache.iotdb.confignode.manager.ConfigManager;
@@ -143,6 +146,9 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.apache.iotdb.confignode.manager.pipe.source.PipeConfigTreePrivilegeParseVisitor.checkGlobalStatus;
+import static org.apache.iotdb.confignode.manager.pipe.source.PipeConfigTreePrivilegeParseVisitor.checkPathsStatus;
+
 public class IoTDBConfigNodeReceiver extends IoTDBFileReceiver {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBConfigNodeReceiver.class);
@@ -157,6 +163,7 @@ public class IoTDBConfigNodeReceiver extends IoTDBFileReceiver {
       new PipeConfigPhysicalPlanExceptionVisitor();
 
   private final ConfigManager configManager = ConfigNode.getInstance().getConfigManager();
+  private final CNAuditLogger auditLogger = configManager.getAuditLogger();
 
   @Override
   public TPipeTransferResp receive(final TPipeTransferReq req) {
@@ -290,61 +297,59 @@ private TSStatus checkPermission(final ConfigPhysicalPlan plan) throws IOExcepti
       return status;
     }
 
+    String database;
+    String templateName;
     switch (plan.getType()) {
       case CreateDatabase:
-        return PathUtils.isTableModelDatabase(((DatabaseSchemaPlan) plan).getSchema().getName())
-            ? configManager
-                .checkUserPrivileges(
-                    username,
-                    new PrivilegeUnion(
-                        ((DatabaseSchemaPlan) plan).getSchema().getName(), PrivilegeType.CREATE))
-                .getStatus()
-            : configManager
-                .checkUserPrivileges(username, new PrivilegeUnion(PrivilegeType.MANAGE_DATABASE))
-                .getStatus();
+        database = ((DatabaseSchemaPlan) plan).getSchema().getName();
+        if (PathUtils.isTableModelDatabase(database)) {
+          status = checkDatabaseStatus(userEntity, PrivilegeType.CREATE, database, false);
+          if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+            return checkGlobalStatus(userEntity, PrivilegeType.SYSTEM, database, true);
+          }
+        }
+        return checkGlobalStatus(userEntity, PrivilegeType.MANAGE_DATABASE, database, true);
       case AlterDatabase:
-        return PathUtils.isTableModelDatabase(((DatabaseSchemaPlan) plan).getSchema().getName())
-            ? configManager
-                .checkUserPrivileges(
-                    username,
-                    new PrivilegeUnion(
-                        ((DatabaseSchemaPlan) plan).getSchema().getName(), PrivilegeType.ALTER))
-                .getStatus()
-            : configManager
-                .checkUserPrivileges(username, new PrivilegeUnion(PrivilegeType.MANAGE_DATABASE))
-                .getStatus();
+        database = ((DatabaseSchemaPlan) plan).getSchema().getName();
+        if (PathUtils.isTableModelDatabase(database)) {
+          status = checkDatabaseStatus(userEntity, PrivilegeType.ALTER, database, false);
+          if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+            return checkGlobalStatus(userEntity, PrivilegeType.SYSTEM, database, true);
+          }
+        }
+        return checkGlobalStatus(userEntity, PrivilegeType.MANAGE_DATABASE, database, true);
       case DeleteDatabase:
-        return PathUtils.isTableModelDatabase(((DeleteDatabasePlan) plan).getName())
-            ? configManager
-                .checkUserPrivileges(
-                    username,
-                    new PrivilegeUnion(((DeleteDatabasePlan) plan).getName(), PrivilegeType.DROP))
-                .getStatus()
-            : configManager
-                .checkUserPrivileges(username, new PrivilegeUnion(PrivilegeType.MANAGE_DATABASE))
-                .getStatus();
+        database = ((DeleteDatabasePlan) plan).getName();
+        if (PathUtils.isTableModelDatabase(database)) {
+          status = checkDatabaseStatus(userEntity, PrivilegeType.DELETE, database, false);
+          if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+            return checkGlobalStatus(userEntity, PrivilegeType.SYSTEM, database, true);
+          }
+        }
+        return checkGlobalStatus(userEntity, PrivilegeType.MANAGE_DATABASE, database, true);
       case ExtendSchemaTemplate:
-        return configManager
-            .checkUserPrivileges(username, new PrivilegeUnion(PrivilegeType.EXTEND_TEMPLATE))
-            .getStatus();
+        return checkGlobalStatus(
+            userEntity,
+            PrivilegeType.EXTEND_TEMPLATE,
+            ((ExtendSchemaTemplatePlan) plan).getTemplateExtendInfo().getTemplateName(),
+            true);
       case CreateSchemaTemplate:
+        templateName = ((CreateSchemaTemplatePlan) plan).getTemplate().getName();
+        return checkGlobalStatus(userEntity, PrivilegeType.SYSTEM, templateName, true);
       case CommitSetSchemaTemplate:
+        templateName = ((CommitSetSchemaTemplatePlan) plan).getName();
+        return checkGlobalStatus(userEntity, PrivilegeType.SYSTEM, templateName, true);
       case PipeUnsetTemplate:
-        return CommonDescriptor.getInstance().getConfig().getDefaultAdminName().equals(username)
-            ? StatusUtils.OK
-            : new TSStatus(TSStatusCode.NO_PERMISSION.getStatusCode())
-                .setMessage("Only the admin user can perform this operation");
+        templateName = ((PipeUnsetSchemaTemplatePlan) plan).getName();
+        return checkGlobalStatus(userEntity, PrivilegeType.SYSTEM, templateName, true);
       case PipeDeleteTimeSeries:
-        return configManager
-            .checkUserPrivileges(
-                username,
-                new PrivilegeUnion(
-                    new ArrayList<>(
-                        PathPatternTree.deserialize(
-                                ((PipeDeleteTimeSeriesPlan) plan).getPatternTreeBytes())
-                            .getAllPathPatterns()),
-                    PrivilegeType.WRITE_SCHEMA))
-            .getStatus();
+        return checkPathsStatus(
+            userEntity,
+            PrivilegeType.WRITE_SCHEMA,
+            new ArrayList<>(
+                PathPatternTree.deserialize(((PipeDeleteTimeSeriesPlan) plan).getPatternTreeBytes())
+                    .getAllPathPatterns()),
+            true);
       case PipeAlterEncodingCompressor:
         // Judge here in the future
         if (configManager
@@ -373,37 +378,30 @@ private TSStatus checkPermission(final ConfigPhysicalPlan plan) throws IOExcepti
                       .serialize());
           return StatusUtils.OK;
         } else {
-          return configManager
-              .checkUserPrivileges(
-                  username,
-                  new PrivilegeUnion(
-                      new ArrayList<>(
-                          PathPatternTree.deserialize(
-                                  ((PipeAlterEncodingCompressorPlan) plan).getPatternTreeBytes())
-                              .getAllPathPatterns()),
-                      PrivilegeType.WRITE_SCHEMA))
-              .getStatus();
+          return checkPathsStatus(
+              userEntity,
+              PrivilegeType.WRITE_SCHEMA,
+              new ArrayList<>(
+                  PathPatternTree.deserialize(
+                          ((PipeAlterEncodingCompressorPlan) plan).getPatternTreeBytes())
+                      .getAllPathPatterns()),
+              true);
         }
       case PipeDeleteLogicalView:
-        return configManager
-            .checkUserPrivileges(
-                username,
-                new PrivilegeUnion(
-                    new ArrayList<>(
-                        PathPatternTree.deserialize(
-                                ((PipeDeleteLogicalViewPlan) plan).getPatternTreeBytes())
-                            .getAllPathPatterns()),
-                    PrivilegeType.WRITE_SCHEMA))
-            .getStatus();
+        return checkPathsStatus(
+            userEntity,
+            PrivilegeType.WRITE_SCHEMA,
+            new ArrayList<>(
+                PathPatternTree.deserialize(
+                        ((PipeDeleteLogicalViewPlan) plan).getPatternTreeBytes())
+                    .getAllPathPatterns()),
+            true);
       case PipeDeactivateTemplate:
-        return configManager
-            .checkUserPrivileges(
-                username,
-                new PrivilegeUnion(
-                    new ArrayList<>(
-                        ((PipeDeactivateTemplatePlan) plan).getTemplateSetInfo().keySet()),
-                    PrivilegeType.WRITE_SCHEMA))
-            .getStatus();
+        return checkPathsStatus(
+            userEntity,
+            PrivilegeType.WRITE_SCHEMA,
+            new ArrayList<>(((PipeDeactivateTemplatePlan) plan).getTemplateSetInfo().keySet()),
+            true);
       case SetTTL:
         return Objects.equals(
                 configManager
@@ -618,6 +616,29 @@ username, new PrivilegeUnion(PrivilegeType.values()[permission], true))
     }
   }
 
+  public static TSStatus checkDatabaseStatus(
+      final IAuditEntity userEntity,
+      final PrivilegeType privilegeType,
+      final String database,
+      final boolean isLastCheck) {
+    final ConfigManager configManager = ConfigNode.getInstance().getConfigManager();
+    final CNAuditLogger logger = configManager.getAuditLogger();
+    final TSStatus result =
+        configManager
+            .getPermissionManager()
+            .checkUserPrivileges(
+                userEntity.getUsername(), new PrivilegeUnion(database, privilegeType))
+            .getStatus();
+    if (result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode() || isLastCheck) {
+      logger.recordAuditLog(
+          userEntity
+              .setPrivilegeType(privilegeType)
+              .setResult(result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()),
+          () -> database);
+    }
+    return result;
+  }
+
   private TSStatus executePlan(final ConfigPhysicalPlan plan) throws ConsensusException {
     final String queryId = generatePseudoQueryId();
     switch (plan.getType()) {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/IoTDBConfigRegionSource.java

@@ -246,15 +246,16 @@ protected Optional<PipeWritePlanEvent> trimRealtimeEventByPrivilege(
         ((PipeConfigRegionWritePlanEvent) event).getConfigPhysicalPlan();
     final Boolean isTableDatabasePlan = isTableDatabasePlan(plan);
     if (!Boolean.TRUE.equals(isTableDatabasePlan)) {
-      final Optional<ConfigPhysicalPlan> result = treePrivilegeParseVisitor.process(plan, userName);
+      final Optional<ConfigPhysicalPlan> result =
+          treePrivilegeParseVisitor.process(plan, userEntity);
       if (result.isPresent()) {
         return Optional.of(
             new PipeConfigRegionWritePlanEvent(result.get(), event.isGeneratedByPipe()));
       }
     }
     if (!Boolean.FALSE.equals(isTableDatabasePlan)) {
       final Optional<ConfigPhysicalPlan> result =
-          TABLE_PRIVILEGE_PARSE_VISITOR.process(plan, userName);
+          TABLE_PRIVILEGE_PARSE_VISITOR.process(plan, userEntity);
       if (result.isPresent()) {
         return Optional.of(
             new PipeConfigRegionWritePlanEvent(result.get(), event.isGeneratedByPipe()));