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,16 @@ public void testSourcePermission() {
       fail(e.getMessage());
     }
   }
+
+  @Test
+  public void testIllegalPassword() {
+    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);
+  }
 }

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-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

@@ -57,6 +57,8 @@
 import org.apache.iotdb.pipe.api.exception.PipeException;
 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 +110,19 @@ 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 PipeException(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/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/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;
     }

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

@@ -543,23 +543,31 @@ public void setAcceptCache(boolean acceptCache) {
   }
 
   @Override
-  public TSStatus checkUser(String username, String password) {
+  public TSStatus checkUser(
+      final String username, final String password, final boolean useEncryptedPassword) {
     checkCacheAvailable();
-    User user = iAuthorCache.getUserCache(username);
+    final User user = iAuthorCache.getUserCache(username);
     if (user != null) {
       if (user.isOpenIdUser()) {
         return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
-      } else if (password != null && AuthUtils.validatePassword(password, user.getPassword())) {
-        return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
-      } else if (password != null
-          && AuthUtils.validatePassword(
-              password, user.getPassword(), AsymmetricEncrypt.DigestAlgorithm.MD5)) {
-        return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+      } else if (password != null) {
+        if (useEncryptedPassword) {
+          return password.equals(user.getPassword())
+              ? RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)
+              : RpcUtils.getStatus(TSStatusCode.WRONG_LOGIN_PASSWORD, "Authentication failed.");
+        } else {
+          return AuthUtils.validatePassword(password, user.getPassword())
+                  || AuthUtils.validatePassword(
+                      password, user.getPassword(), AsymmetricEncrypt.DigestAlgorithm.MD5)
+              ? RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)
+              : RpcUtils.getStatus(TSStatusCode.WRONG_LOGIN_PASSWORD, "Authentication failed.");
+        }
       } else {
         return RpcUtils.getStatus(TSStatusCode.WRONG_LOGIN_PASSWORD, "Authentication failed.");
       }
     } else {
-      TLoginReq req = new TLoginReq(username, password);
+      TLoginReq req =
+          new TLoginReq(username, password).setUseEncryptedPassword(useEncryptedPassword);
       TPermissionInfoResp status = null;
       try (ConfigNodeClient configNodeClient =
           CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {

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

@@ -36,7 +36,8 @@
 
 public interface IAuthorityFetcher {
 
-  TSStatus checkUser(String username, String password);
+  TSStatus checkUser(
+      final String username, final String password, final boolean useEncryptedPassword);
 
   boolean checkRole(String username, String roleName);
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java

@@ -953,7 +953,7 @@ protected TSStatus login() {
     }
 
     long userId = AuthorityChecker.getUserId(username).orElse(-1L);
-    Long timeToExpire = DataNodeAuthUtils.checkPasswordExpiration(userId, password);
+    Long timeToExpire = DataNodeAuthUtils.checkPasswordExpiration(userId, password, false);
     if (timeToExpire != null && timeToExpire <= System.currentTimeMillis()) {
       return RpcUtils.getStatus(
           TSStatusCode.ILLEGAL_PASSWORD.getStatusCode(),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/writeback/WriteBackSink.java

@@ -83,6 +83,7 @@
 import java.util.stream.Collectors;
 
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_CLI_HOSTNAME;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SKIP_IF_NO_PRIVILEGES;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USER_ID;
@@ -91,6 +92,7 @@
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_USE_EVENT_USER_NAME_DEFAULT_VALUE;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_USE_EVENT_USER_NAME_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_CLI_HOSTNAME;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_USERNAME_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_USER_ID;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_USER_KEY;
@@ -156,6 +158,8 @@ public void customize(
             SINK_IOTDB_USER_KEY,
             CONNECTOR_IOTDB_USERNAME_KEY,
             SINK_IOTDB_USERNAME_KEY);
+    String passwordString =
+        parameters.getStringByKeys(CONNECTOR_IOTDB_PASSWORD_KEY, SINK_IOTDB_PASSWORD_KEY);
     String cliHostnameString =
         parameters.getStringByKeys(CONNECTOR_IOTDB_CLI_HOSTNAME, SINK_IOTDB_CLI_HOSTNAME);
     userEntity = new UserEntity(Long.parseLong(userIdString), usernameString, cliHostnameString);
@@ -191,6 +195,24 @@ public void customize(
     if (SESSION_MANAGER.getCurrSession() == null) {
       SESSION_MANAGER.registerSession(session);
     }
+
+    // Check the password and its expiration
+    if (Objects.nonNull(passwordString)
+        && SESSION_MANAGER
+                .login(
+                    session,
+                    usernameString,
+                    passwordString,
+                    ZoneId.systemDefault().toString(),
+                    SessionManager.CURRENT_RPC_VERSION,
+                    IoTDBConstant.ClientVersion.V_1_0,
+                    IClientSession.SqlDialect.TREE,
+                    environment.getRegionId() >= 0)
+                .getCode()
+            != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+      throw new PipeException(
+          String.format("Failed to check password for pipe %s.", environment.getPipeName()));
+    }
   }
 
   @Override