Pipe: Fixed the auth implementation error (#16847)

* fix

* fix

* non-geneic

* fix

Author: Caideyipi

integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipePermissionIT.java

@@ -19,8 +19,6 @@
 
 package org.apache.iotdb.pipe.it.single;
 
-import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
-import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
 import org.apache.iotdb.db.it.utils.TestUtils;
 import org.apache.iotdb.it.framework.IoTDBTestRunner;
 import org.apache.iotdb.itbase.category.MultiClusterIT1;
@@ -33,6 +31,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;
@@ -44,7 +43,7 @@
 public class IoTDBPipePermissionIT extends AbstractPipeSingleIT {
   @Test
   public void testSinkPermission() {
-    TestUtils.executeNonQuery(env, "create user `thulab` 'passwd'", null);
+    TestUtils.executeNonQuery(env, "create user `thulab` 'StrngPsWd@623451'", null);
 
     // Shall fail if username is specified without password
     try (final Connection connection = env.getConnection(BaseEnv.TABLE_SQL_DIALECT);
@@ -90,7 +89,8 @@ public void testSinkPermission() {
     // Successfully alter
     try (final Connection connection = env.getConnection(BaseEnv.TABLE_SQL_DIALECT);
         final Statement statement = connection.createStatement()) {
-      statement.execute("alter pipe a2b modify sink ('username'='thulab', 'password'='passwd')");
+      statement.execute(
+          "alter pipe a2b modify sink ('username'='thulab', 'password'='StrngPsWd@623451')");
     } catch (final SQLException e) {
       e.printStackTrace();
       fail("Alter pipe shall not fail if user and password are specified");
@@ -156,14 +156,12 @@ public void testSinkPermission() {
     }
 
     // A user shall only see its own pipe
-    try (final SyncConfigNodeIServiceClient client =
-        (SyncConfigNodeIServiceClient) env.getLeaderConfigNodeConnection()) {
-      Assert.assertEquals(
-          1,
-          client
-              .showPipe(new TShowPipeReq().setIsTableModel(true).setUserName("thulab"))
-              .pipeInfoList
-              .size());
+    try (final Connection connection =
+            env.getConnection("thulab", "StrngPsWd@623451", BaseEnv.TABLE_SQL_DIALECT);
+        final Statement statement = connection.createStatement()) {
+      final ResultSet result = statement.executeQuery("show pipes");
+      Assert.assertTrue(result.next());
+      Assert.assertFalse(result.next());
     } catch (Exception e) {
       fail(e.getMessage());
     }
@@ -181,7 +179,8 @@ public void testSinkPermissionWithHistoricalDataAndTablePattern() {
         BaseEnv.TABLE_SQL_DIALECT,
         env,
         Arrays.asList(
-            "create user thulab 'passwD@123456'", "grant INSERT on test.test1 to user thulab"),
+            "create user thulab 'StrngPsWd@623451@123456'",
+            "grant INSERT on test.test1 to user thulab"),
         null);
 
     // Write some data
@@ -196,7 +195,7 @@ public void testSinkPermissionWithHistoricalDataAndTablePattern() {
           "create pipe a2b "
               + "with source ('database'='test1', 'table'='test1') "
               + "with processor('processor'='rename-database-processor', 'processor.new-db-name'='test') "
-              + "with sink ('sink'='write-back-sink', 'username'='thulab', 'password'='passwD@123456')");
+              + "with sink ('sink'='write-back-sink', 'username'='thulab', 'password'='StrngPsWd@623451@123456')");
     } catch (final SQLException e) {
       e.printStackTrace();
       fail("Create pipe without user shall succeed if use the current session");

integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java

@@ -575,6 +575,8 @@ public void testInformationSchema() throws SQLException {
       Assert.assertThrows(
           SQLException.class, () -> statement.execute("select * from config_nodes"));
       Assert.assertThrows(SQLException.class, () -> statement.execute("select * from data_nodes"));
+      Assert.assertThrows(
+          SQLException.class, () -> statement.executeQuery("select * from pipe_plugins"));
 
       // Filter out not self-created pipes
       TestUtils.assertResultSetEqual(
@@ -583,12 +585,6 @@ public void testInformationSchema() throws SQLException {
           Collections.emptySet());
 
       // No auth needed
-      TestUtils.assertResultSetEqual(
-          statement.executeQuery(
-              "select * from pipe_plugins where plugin_name = 'IOTDB-THRIFT-SINK'"),
-          "plugin_name,plugin_type,class_name,plugin_jar,",
-          Collections.singleton(
-              "IOTDB-THRIFT-SINK,Builtin,org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.thrift.IoTDBThriftSink,null,"));
 
       TestUtils.assertResultSetEqual(
           statement.executeQuery(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java

@@ -28,6 +28,7 @@
 import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
 import org.apache.iotdb.common.rpc.thrift.TEndPoint;
 import org.apache.iotdb.commons.audit.UserEntity;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
 import org.apache.iotdb.commons.conf.IoTDBConstant;
 import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
 import org.apache.iotdb.commons.exception.auth.AccessDeniedException;
@@ -82,6 +83,7 @@
 import org.apache.iotdb.db.utils.TimestampPrecisionUtils;
 import org.apache.iotdb.rpc.TSStatusCode;
 
+import org.apache.thrift.TException;
 import org.apache.tsfile.block.column.ColumnBuilder;
 import org.apache.tsfile.common.conf.TSFileConfig;
 import org.apache.tsfile.enums.TSDataType;
@@ -143,7 +145,7 @@ public static Iterator<TsBlock> getSupplier(
         case InformationSchema.PIPES:
           return new PipeSupplier(dataTypes, userEntity.getUsername());
         case InformationSchema.PIPE_PLUGINS:
-          return new PipePluginSupplier(dataTypes);
+          return new PipePluginSupplier(dataTypes, userEntity);
         case InformationSchema.TOPICS:
           return new TopicSupplier(dataTypes, userEntity);
         case InformationSchema.SUBSCRIPTIONS:
@@ -603,8 +605,10 @@ public boolean hasNext() {
   private static class PipePluginSupplier extends TsBlockSupplier {
     private final Iterator<PipePluginMeta> iterator;
 
-    private PipePluginSupplier(final List<TSDataType> dataTypes) throws Exception {
+    private PipePluginSupplier(final List<TSDataType> dataTypes, final UserEntity entity)
+        throws ClientManagerException, TException {
       super(dataTypes);
+      accessControl.checkUserGlobalSysPrivilege(entity);
       try (final ConfigNodeClient client =
           ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
         iterator =

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java

@@ -1270,7 +1270,6 @@ protected IConfigTask visitStopPipe(StopPipe node, MPPQueryContext context) {
   @Override
   protected IConfigTask visitShowPipes(ShowPipes node, MPPQueryContext context) {
     context.setQueryType(QueryType.READ);
-    accessControl.checkUserGlobalSysPrivilege(context);
     return new ShowPipeTask(node, context.getSession().getUserName());
   }