Fix account unlock propagation for pipe password checks

Author: Caideyipi

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

@@ -595,6 +595,7 @@ public void testIllegalPassword() throws Exception {
 
     try {
       statement.execute("alter pipe a2b modify source ('password'='fake')");
+      fail();
     } catch (final SQLException e) {
       Assert.assertEquals("801: Failed to check password for pipe a2b.", e.getMessage());
     }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java

@@ -311,6 +311,7 @@ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOExcept
         case UpdateUserV2:
         case CreateUserWithRawPassword:
         case RenameUser:
+        case AccountUnlock:
           plan = new AuthorTreePlan(configPhysicalPlanType);
           break;
         case RCreateUser:
@@ -343,6 +344,7 @@ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOExcept
         case RRevokeUserSysPri:
         case RRevokeRoleSysPri:
         case RRenameUser:
+        case RAccountUnlock:
           plan = new AuthorRelationalPlan(configPhysicalPlanType);
           break;
         case ApplyConfigNode:

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java

@@ -333,6 +333,7 @@ public enum ConfigPhysicalPlanType {
   RDropUserV2((short) 2103),
   RenameUser((short) 2104),
   RRenameUser((short) 2105),
+  AccountUnlock((short) 2106),
 
   EnableSeparationOfAdminPowers((short) 2200),
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanVisitor.java

@@ -104,6 +104,8 @@ public R process(final ConfigPhysicalPlan plan, final C context) {
         return visitGrantRoleToUser((AuthorTreePlan) plan, context);
       case RevokeRoleFromUser:
         return visitRevokeRoleFromUser((AuthorTreePlan) plan, context);
+      case AccountUnlock:
+        return visitAccountUnlock((AuthorTreePlan) plan, context);
       case RCreateUser:
         return visitRCreateUser((AuthorRelationalPlan) plan, context);
       case RCreateRole:
@@ -160,6 +162,8 @@ public R process(final ConfigPhysicalPlan plan, final C context) {
         return visitRRevokeUserSysPrivilege((AuthorRelationalPlan) plan, context);
       case RRevokeRoleSysPri:
         return visitRRevokeRoleSysPrivilege((AuthorRelationalPlan) plan, context);
+      case RAccountUnlock:
+        return visitRAccountUnlock((AuthorRelationalPlan) plan, context);
       case SetTTL:
         return visitTTL((SetTTLPlan) plan, context);
       case PipeCreateTableOrView:
@@ -310,6 +314,10 @@ public R visitRevokeRoleFromUser(final AuthorTreePlan revokeRoleFromUserPlan, fi
     return visitPlan(revokeRoleFromUserPlan, context);
   }
 
+  public R visitAccountUnlock(final AuthorTreePlan accountUnlockPlan, final C context) {
+    return visitPlan(accountUnlockPlan, context);
+  }
+
   public R visitRCreateUser(final AuthorRelationalPlan rCreateUserPlan, final C context) {
     return visitPlan(rCreateUserPlan, context);
   }
@@ -426,6 +434,10 @@ public R visitRRevokeRoleSysPrivilege(
     return visitPlan(rRevokeRoleSysPrivilegePlan, context);
   }
 
+  public R visitRAccountUnlock(final AuthorRelationalPlan rAccountUnlockPlan, final C context) {
+    return visitPlan(rAccountUnlockPlan, context);
+  }
+
   public R visitTTL(final SetTTLPlan setTTLPlan, final C context) {
     return visitPlan(setTTLPlan, context);
   }

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

@@ -74,7 +74,9 @@ public static ConfigPhysicalPlanType getConfigPhysicalPlanTypeFromAuthorType(int
       throw new IndexOutOfBoundsException(ConfigNodeMessages.INVALID_AUTHOR_TYPE_ORDINAL);
     }
     ConfigPhysicalPlanType configPhysicalPlanType;
-    if (authorType >= AuthorType.RENAME_USER.ordinal()) {
+    if (authorType == AuthorType.ACCOUNT_UNLOCK.ordinal()) {
+      return ConfigPhysicalPlanType.AccountUnlock;
+    } else if (authorType >= AuthorType.RENAME_USER.ordinal()) {
       AuthorType type = AuthorType.values()[authorType];
       switch (type) {
         case RENAME_USER:
@@ -105,6 +107,8 @@ public static ConfigPhysicalPlanType getConfigPhysicalPlanTypeFromAuthorRType(in
     ConfigPhysicalPlanType configPhysicalPlanType;
     if (authorRType == AuthorRType.RENAME_USER.ordinal()) {
       configPhysicalPlanType = ConfigPhysicalPlanType.RRenameUser;
+    } else if (authorRType == AuthorRType.ACCOUNT_UNLOCK.ordinal()) {
+      configPhysicalPlanType = ConfigPhysicalPlanType.RAccountUnlock;
     } else {
       configPhysicalPlanType =
           ConfigPhysicalPlanType.values()[

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

@@ -124,6 +124,9 @@ public TSStatus executeAuthorNonQuery(AuthorTreePlan authorPlan) {
         case RenameUser:
           authorizer.renameUser(userName, newUsername);
           break;
+        case AccountUnlock:
+          authorizer.getUser(userName);
+          break;
         case CreateUser:
           authorizer.createUser(userName, password);
           break;
@@ -241,6 +244,7 @@ public TSStatus executeRelationalAuthorNonQuery(AuthorRelationalPlan authorPlan)
           authorizer.renameUser(userName, newUsername);
           break;
         case RAccountUnlock:
+          authorizer.getUser(userName);
           break;
         case RDropRole:
           authorizer.deleteRole(roleName);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java

@@ -488,6 +488,7 @@ public TSStatus executeNonQueryPlan(ConfigPhysicalPlan physicalPlan)
       case RevokeRoleFromUserDep:
       case UpdateUserDep:
       case RenameUser:
+      case AccountUnlock:
       case RCreateRole:
       case RCreateUser:
       case RDropUser:

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java

@@ -28,6 +28,7 @@
 import org.apache.iotdb.confignode.client.sync.CnToDnSyncRequestType;
 import org.apache.iotdb.confignode.client.sync.SyncDataNodeClientPool;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
 import org.apache.iotdb.confignode.consensus.request.write.auth.AuthorPlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
 import org.apache.iotdb.confignode.i18n.ProcedureMessages;
@@ -100,6 +101,11 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, AuthOperationProcedu
           TSStatus status;
           req.setUsername(user);
           req.setRoleName(role);
+          if (plan.getAuthorType() == ConfigPhysicalPlanType.AccountUnlock
+              || plan.getAuthorType() == ConfigPhysicalPlanType.RAccountUnlock) {
+            // For account unlock, role carries the optional login address.
+            req.setNeedDisconnect(true);
+          }
           Iterator<Pair<TDataNodeConfiguration, Long>> it = dataNodesToInvalid.iterator();
           while (it.hasNext()) {
             Pair<TDataNodeConfiguration, Long> pair = it.next();

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedureTest.java

@@ -98,6 +98,34 @@ public void serializeDeserializeTest() throws IOException {
       fail();
     }
 
+    try {
+      final AuthOperationProcedure proc =
+          new AuthOperationProcedure(
+              new AuthorTreePlan(
+                  ConfigPhysicalPlanType.AccountUnlock,
+                  "user1",
+                  "",
+                  "",
+                  "",
+                  Collections.emptySet(),
+                  false,
+                  Collections.emptyList()),
+              datanodes,
+              false);
+      proc.serialize(outputStream);
+      final ByteBuffer buffer =
+          ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
+
+      final AuthOperationProcedure proc2 =
+          (AuthOperationProcedure) ProcedureFactory.getInstance().create(buffer);
+      Assert.assertEquals(proc, proc2);
+      buffer.clear();
+      byteArrayOutputStream.reset();
+    } catch (final Exception e) {
+      e.printStackTrace();
+      fail();
+    }
+
     try {
       final int begin = ConfigPhysicalPlanType.RCreateUser.ordinal();
       final int end = ConfigPhysicalPlanType.RRevokeRoleSysPri.ordinal();
@@ -129,5 +157,33 @@ public void serializeDeserializeTest() throws IOException {
       e.printStackTrace();
       fail();
     }
+
+    try {
+      final AuthOperationProcedure proc =
+          new AuthOperationProcedure(
+              new AuthorRelationalPlan(
+                  ConfigPhysicalPlanType.RAccountUnlock,
+                  "user1",
+                  "127.0.0.1",
+                  "",
+                  "",
+                  Collections.emptySet(),
+                  false,
+                  ""),
+              datanodes,
+              false);
+      proc.serialize(outputStream);
+      final ByteBuffer buffer =
+          ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
+
+      final AuthOperationProcedure proc2 =
+          (AuthOperationProcedure) ProcedureFactory.getInstance().create(buffer);
+      Assert.assertEquals(proc, proc2);
+      buffer.clear();
+      byteArrayOutputStream.reset();
+    } catch (final Exception e) {
+      e.printStackTrace();
+      fail();
+    }
   }
 }

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

@@ -424,44 +424,31 @@ private void onOperatePermissionSuccess(Object plan) {
 
   @Override
   public SettableFuture<ConfigTaskResult> operatePermission(AuthorStatement authorStatement) {
-    return handleAccountUnlock(
-        authorStatement,
-        authorStatement.getUserName(),
-        false,
-        () -> onOperatePermissionSuccess(authorStatement));
+    return handleAccountUnlock(authorStatement, false);
   }
 
   @Override
   public SettableFuture<ConfigTaskResult> operatePermission(
       RelationalAuthorStatement authorStatement) {
-    return handleAccountUnlock(
-        authorStatement,
-        authorStatement.getUserName(),
-        true,
-        () -> onOperatePermissionSuccess(authorStatement));
+    return handleAccountUnlock(authorStatement, true);
   }
 
   private SettableFuture<ConfigTaskResult> handleAccountUnlock(
-      Object authorStatement, String username, boolean isRelational, Runnable successCallback) {
+      Object authorStatement, boolean isRelational) {
 
     if (isUnlockStatement(authorStatement, isRelational)) {
-      final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
-      final User user;
-      try {
-        user = getUser(username, false);
-      } catch (final IoTDBRuntimeException e) {
-        future.setException(e);
-        return future;
-      }
       String loginAddr =
           isRelational
               ? ((RelationalAuthorStatement) authorStatement).getLoginAddr()
               : ((AuthorStatement) authorStatement).getLoginAddr();
 
-      LoginLockManager.getInstance().unlock(user.getUserId(), loginAddr);
-      successCallback.run();
-      future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
-      return future;
+      // Reuse roleName to carry the optional login address for the internal unlock broadcast.
+      if (isRelational) {
+        ((RelationalAuthorStatement) authorStatement).setRoleName(loginAddr);
+      } else {
+        ((AuthorStatement) authorStatement).setRoleName(loginAddr);
+      }
+      return operatePermissionInternal(authorStatement, isRelational);
     }
     return operatePermissionInternal(authorStatement, isRelational);
   }
@@ -748,7 +735,7 @@ public Role cacheRole(String roleName, TPermissionInfoResp tPermissionInfoResp)
 
   private TAuthorizerReq statementToAuthorizerReq(AuthorStatement authorStatement)
       throws AuthException {
-    if (authorStatement.getAuthorType() == null) {
+    if (authorStatement.getNodeNameList() == null) {
       authorStatement.setNodeNameList(new ArrayList<>());
     }
     return new TAuthorizerReq(