Pipe: Implemented the login check in customize method & Added mosaic to subtask string in logger & Optimized the error code of illegal pipe / pipe plugin name string

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

@@ -506,4 +506,129 @@ public void testSourcePermission() {
       fail(e.getMessage());
     }
   }
+
+  @Test
+  public void testIllegalPassword() throws Exception {
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "create user `thulab` 'passwd123456'",
+            "create role `admin`",
+            "grant role `admin` to `thulab`",
+            "grant WRITE, READ, SYSTEM, SECURITY on root.** to role `admin`"),
+        null);
+
+    TestUtils.executeNonQuery(
+        senderEnv,
+        "create aligned timeSeries root.vehicle.plane(temperature DOUBLE, pressure INT32)");
+    TestUtils.executeNonQuery(
+        receiverEnv,
+        "create aligned timeSeries root.vehicle.plane(temperature DOUBLE, pressure INT32)");
+
+    Connection connection = senderEnv.getConnection();
+    Statement statement = connection.createStatement();
+    try {
+      statement.execute(
+          String.format(
+              "create pipe a2b"
+                  + " with source ("
+                  + "'user'='thulab'"
+                  + ", 'password'='passwd')"
+                  + " with sink ("
+                  + "'node-urls'='%s')",
+              receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString()));
+      fail();
+    } catch (final Exception e) {
+      Assert.assertEquals("801: Failed to check password for pipe a2b.", e.getMessage());
+    }
+
+    try {
+      statement.execute(
+          "create pipe a2b ('sink'='write-back-sink', 'user'='thulab', 'password'='passwd')");
+      fail();
+    } catch (final Exception e) {
+      Assert.assertEquals("801: Failed to check password for pipe a2b.", e.getMessage());
+    }
+
+    statement.execute(
+        String.format(
+            "create pipe a2b"
+                + " with source ("
+                + "'user'='thulab'"
+                + ", 'password'='passwd123456')"
+                + " with sink ("
+                + "'node-urls'='%s')",
+            receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString()));
+
+    TestUtils.executeNonQuery(
+        senderEnv, "insert into root.vehicle.plane(temperature, pressure) values (36.5, 1103)");
+
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "select count(pressure) from root.vehicle.plane",
+        "count(root.vehicle.plane.pressure),",
+        Collections.singleton("1,"));
+
+    statement.execute("alter user thulab set password 'newPassword'");
+
+    try {
+      TestUtils.restartCluster(senderEnv);
+    } catch (final Throwable e) {
+      e.printStackTrace();
+      return;
+    }
+
+    connection = senderEnv.getConnection();
+    statement = connection.createStatement();
+    TestUtils.executeNonQuery(
+        senderEnv, "insert into root.vehicle.plane(temperature, pressure) values (36.5, 1103)");
+
+    TestUtils.assertDataAlwaysOnEnv(
+        receiverEnv,
+        "select count(pressure) from root.vehicle.plane",
+        "count(root.vehicle.plane.pressure),",
+        Collections.singleton("1,"));
+
+    try {
+      statement.execute("alter pipe a2b modify source ('password'='fake')");
+    } catch (final SQLException e) {
+      Assert.assertEquals("801: Failed to check password for pipe a2b.", e.getMessage());
+    }
+
+    statement.execute("alter pipe a2b modify source ('password'='newPassword')");
+
+    // Test empty alter
+    statement.execute("alter pipe a2b");
+
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "select count(pressure) from root.vehicle.plane",
+        "count(root.vehicle.plane.pressure),",
+        Collections.singleton("2,"));
+
+    statement.execute("alter user thulab set password 'anotherPassword'");
+
+    try {
+      TestUtils.restartCluster(senderEnv);
+    } catch (final Throwable e) {
+      e.printStackTrace();
+      return;
+    }
+
+    connection = senderEnv.getConnection();
+    statement = connection.createStatement();
+    TestUtils.executeNonQuery(
+        senderEnv, "insert into root.vehicle.plane(temperature, pressure) values (36.5, 1103)");
+    statement.execute("alter user thulab set password 'newPassword'");
+    statement.execute("alter pipe a2b");
+
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "select count(pressure) from root.vehicle.plane",
+        "count(root.vehicle.plane.pressure),",
+        Collections.singleton("3,"));
+
+    statement.close();
+    connection.close();
+  }
 }

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java

@@ -371,7 +371,7 @@ public PipeParameters addOrReplaceEquivalentAttributesWithClone(final PipeParame
     return new PipeParameters(thisMap);
   }
 
-  private static class KeyReducer {
+  public static class KeyReducer {
 
     private static final Set<String> FIRST_PREFIXES = new HashSet<>();
     private static final Set<String> SECOND_PREFIXES = new HashSet<>();
@@ -399,7 +399,7 @@ static String shallowReduce(String key) {
       return key;
     }
 
-    static String reduce(String key) {
+    public static String reduce(String key) {
       if (key == null) {
         return null;
       }

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/exception/PipePasswordCheckException.java

@@ -0,0 +1,26 @@
+/*
+ * 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.iotdb.pipe.api.exception;
+
+public class PipePasswordCheckException extends PipeException {
+  public PipePasswordCheckException(final String message) {
+    super(message);
+  }
+}

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -1332,10 +1332,11 @@ public DataSet queryPermission(final AuthorPlan authorPlan) {
   }
 
   @Override
-  public TPermissionInfoResp login(String username, String password) {
+  public TPermissionInfoResp login(
+      final String username, final String password, final boolean useEncryptedPassword) {
     TSStatus status = confirmLeader();
     if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-      return permissionManager.login(username, password);
+      return permissionManager.login(username, password, useEncryptedPassword);
     } else {
       TPermissionInfoResp resp = AuthUtils.generateEmptyPermissionInfoResp();
       resp.setStatus(status);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java

@@ -501,7 +501,8 @@ TDataPartitionTableResp getOrCreateDataPartition(
   DataSet queryPermission(final AuthorPlan authorPlan);
 
   /** login. */
-  TPermissionInfoResp login(String username, String password);
+  TPermissionInfoResp login(
+      final String username, final String password, final boolean useEncryptedPassword);
 
   /** Check User Privileges. */
   TPermissionInfoResp checkUserPrivileges(String username, PrivilegeUnion union);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/PermissionManager.java

@@ -112,8 +112,9 @@ protected ConsensusManager getConsensusManager() {
     return configManager.getConsensusManager();
   }
 
-  public TPermissionInfoResp login(String username, String password) {
-    return authorInfo.login(username, password);
+  public TPermissionInfoResp login(
+      final String username, final String password, final boolean useEncryptedPassword) {
+    return authorInfo.login(username, password, useEncryptedPassword);
   }
 
   public String login4Pipe(final String userName, final String password) {

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

@@ -1225,7 +1225,7 @@ protected boolean shouldLogin() {
 
   @Override
   protected TSStatus login() {
-    return configManager.login(username, password).getStatus();
+    return configManager.login(username, password, false).getStatus();
   }
 
   @Override

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

@@ -55,8 +55,11 @@
 import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
 import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
 import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.apache.iotdb.pipe.api.exception.PipePasswordCheckException;
 import org.apache.iotdb.rpc.TSStatusCode;
 
+import javax.annotation.Nonnull;
+
 import java.io.IOException;
 import java.nio.file.Paths;
 import java.util.Collections;
@@ -108,6 +111,20 @@ public void customize(
     PipeConfigNodeRemainingTimeMetrics.getInstance().register(this);
   }
 
+  @Override
+  protected void login(final @Nonnull String password) {
+    if (ConfigNode.getInstance()
+            .getConfigManager()
+            .getPermissionManager()
+            .login(userName, password, true)
+            .getStatus()
+            .getCode()
+        != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+      throw new PipePasswordCheckException(
+          String.format("Failed to check password for pipe %s.", pipeName));
+    }
+  }
+
   @Override
   protected AbstractPipeListeningQueue getListeningQueue() {
     return PipeConfigNodeAgent.runtime().listener();

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/auth/AuthorInfo.java

@@ -124,8 +124,9 @@ public void setAuthorQueryPlanExecutor(IAuthorPlanExecutor authorPlanExecutor) {
     this.authorPlanExecutor = authorPlanExecutor;
   }
 
-  public TPermissionInfoResp login(String username, String password) {
-    return authorPlanExecutor.login(username, password);
+  public TPermissionInfoResp login(
+      final String username, final String password, final boolean useEncryptedPassword) {
+    return authorPlanExecutor.login(username, password, useEncryptedPassword);
   }
 
   public String login4Pipe(final String username, final String password) {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/auth/AuthorPlanExecutor.java

@@ -74,13 +74,14 @@ public AuthorPlanExecutor(IAuthorizer authorizer) {
   }
 
   @Override
-  public TPermissionInfoResp login(String username, String password) {
+  public TPermissionInfoResp login(
+      String username, final String password, final boolean useEncryptedPassword) {
     boolean status;
     String loginMessage = null;
     TSStatus tsStatus = new TSStatus();
     TPermissionInfoResp result = new TPermissionInfoResp();
     try {
-      status = authorizer.login(username, password);
+      status = authorizer.login(username, password, useEncryptedPassword);
       if (status) {
         // Bring this user's permission information back to the datanode for caching
         if (authorizer instanceof OpenIdAuthorizer) {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/auth/IAuthorPlanExecutor.java

@@ -33,7 +33,8 @@
 import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
 
 public interface IAuthorPlanExecutor {
-  TPermissionInfoResp login(String username, String password);
+  TPermissionInfoResp login(
+      final String username, final String password, final boolean useEncryptedPassword);
 
   String login4Pipe(final String username, final String password);
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java

@@ -107,14 +107,18 @@ public boolean executeFromValidateTask(final ConfigNodeProcedureEnv env) throws
                 PipeSourceConstant.EXTRACTOR_IOTDB_USER_KEY,
                 PipeSourceConstant.SOURCE_IOTDB_USER_KEY,
                 PipeSourceConstant.EXTRACTOR_IOTDB_USERNAME_KEY,
-                PipeSourceConstant.SOURCE_IOTDB_USERNAME_KEY);
+                PipeSourceConstant.SOURCE_IOTDB_USERNAME_KEY,
+                PipeSourceConstant.EXTRACTOR_IOTDB_PASSWORD_KEY,
+                PipeSourceConstant.SOURCE_IOTDB_PASSWORD_KEY);
     final boolean checkSink =
         new PipeParameters(alterPipeRequest.getConnectorAttributes())
             .hasAnyAttributes(
                 PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY,
                 PipeSinkConstant.SINK_IOTDB_USER_KEY,
                 PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY,
-                PipeSinkConstant.SINK_IOTDB_USERNAME_KEY);
+                PipeSinkConstant.SINK_IOTDB_USERNAME_KEY,
+                PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY,
+                PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY);
 
     pipeTaskInfo.get().checkAndUpdateRequestBeforeAlterPipe(alterPipeRequest);
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java

@@ -174,7 +174,9 @@ public static void checkAndEnrichSourceAuthentication(
     if (sourceParameters.hasAttribute(PipeSourceConstant.EXTRACTOR_IOTDB_USER_KEY)
         || sourceParameters.hasAttribute(PipeSourceConstant.SOURCE_IOTDB_USER_KEY)
         || sourceParameters.hasAttribute(PipeSourceConstant.EXTRACTOR_IOTDB_USERNAME_KEY)
-        || sourceParameters.hasAttribute(PipeSourceConstant.SOURCE_IOTDB_USERNAME_KEY)) {
+        || sourceParameters.hasAttribute(PipeSourceConstant.SOURCE_IOTDB_USERNAME_KEY)
+        || sourceParameters.hasAttribute(PipeSourceConstant.EXTRACTOR_IOTDB_PASSWORD_KEY)
+        || sourceParameters.hasAttribute(PipeSourceConstant.SOURCE_IOTDB_PASSWORD_KEY)) {
       final String hashedPassword =
           env.getConfigManager()
               .getPermissionManager()
@@ -216,7 +218,9 @@ public static void checkAndEnrichSinkAuthentication(
     if (sinkParameters.hasAttribute(PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY)
         || sinkParameters.hasAttribute(PipeSinkConstant.SINK_IOTDB_USER_KEY)
         || sinkParameters.hasAttribute(PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY)
-        || sinkParameters.hasAttribute(PipeSinkConstant.SINK_IOTDB_USERNAME_KEY)) {
+        || sinkParameters.hasAttribute(PipeSinkConstant.SINK_IOTDB_USERNAME_KEY)
+        || sinkParameters.hasAttribute(PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY)
+        || sinkParameters.hasAttribute(PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY)) {
       final String hashedPassword =
           env.getConfigManager()
               .getPermissionManager()

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java

@@ -744,7 +744,10 @@ public TAuthorizerResp queryRPermission(final TAuthorizerRelationalReq req) {
 
   @Override
   public TPermissionInfoResp login(TLoginReq req) {
-    return configManager.login(req.getUserrname(), req.getPassword());
+    return configManager.login(
+        req.getUserrname(),
+        req.getPassword(),
+        req.isSetUseEncryptedPassword() && req.isUseEncryptedPassword());
   }
 
   @Override

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/AuthorInfoTest.java

@@ -664,7 +664,7 @@ public void createUserWithRawPassword() {
             new ArrayList<>());
     status = authorInfo.authorNonQuery(authorPlan);
     assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
-    TPermissionInfoResp result = authorInfo.login("testuser", "password123456");
+    TPermissionInfoResp result = authorInfo.login("testuser", "password123456", false);
     assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), result.getStatus().getCode());
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java

@@ -142,8 +142,14 @@ public static Optional<Long> getUserId(String username) {
     return Optional.ofNullable(user == null ? null : user.getUserId());
   }
 
-  public static TSStatus checkUser(String userName, String password) {
-    TSStatus status = authorityFetcher.get().checkUser(userName, password);
+  public static TSStatus checkUser(final String userName, final String password) {
+    return checkUser(userName, password, false);
+  }
+
+  public static TSStatus checkUser(
+      final String userName, final String password, final boolean useEncryptedPassword) {
+    final TSStatus status =
+        authorityFetcher.get().checkUser(userName, password, useEncryptedPassword);
     if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
       return status;
     }