Fix account unlock propagation for pipe password checks (#17814)

Caideyipi · JackieTien97 · commit 540c00891345 · 2026-06-29T13:43:01.000+08:00
* Fix account unlock propagation for pipe password checks

* Fix account unlock user validation
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java b/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());
     }
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java b/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:
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java b/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),
 
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanVisitor.java b/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);
   }
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/auth/AuthorInfo.java b/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()[
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/auth/AuthorPlanExecutor.java b/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:
+          checkUserExistsForAccountUnlock(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:
+          checkUserExistsForAccountUnlock(userName);
           break;
         case RDropRole:
           authorizer.deleteRole(roleName);
@@ -448,6 +452,14 @@ public TSStatus executeRelationalAuthorNonQuery(AuthorRelationalPlan authorPlan)
     return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
   }
 
+  private void checkUserExistsForAccountUnlock(final String userName) throws AuthException {
+    // Account unlock has no persistent ConfigNode auth state change, but the write path needs this
+    // validation before broadcasting DataNode unlocks and propagating through pipe.
+    if (authorizer.getUser(userName) == null) {
+      throw new AuthException(TSStatusCode.USER_NOT_EXIST, NO_USER_MSG + userName);
+    }
+  }
+
   @Override
   public PermissionInfoResp executeListUsers(final AuthorPlan plan) throws AuthException {
     final PermissionInfoResp result = new PermissionInfoResp();
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java b/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:
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java b/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();
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/auth/AuthorPlanExecutorTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/auth/AuthorPlanExecutorTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.confignode.persistence.auth;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.auth.authorizer.IAuthorizer;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+import org.apache.iotdb.confignode.consensus.request.write.auth.AuthorRelationalPlan;
+import org.apache.iotdb.confignode.consensus.request.write.auth.AuthorTreePlan;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class AuthorPlanExecutorTest {
+
+  @Test
+  public void testAccountUnlockRequiresExistingUser() throws Exception {
+    final IAuthorizer authorizer = mock(IAuthorizer.class);
+    when(authorizer.getUser("missing")).thenReturn(null);
+
+    final AuthorPlanExecutor executor = new AuthorPlanExecutor(authorizer);
+    final TSStatus status =
+        executor.executeAuthorNonQuery(
+            new AuthorTreePlan(
+                ConfigPhysicalPlanType.AccountUnlock,
+                "missing",
+                "",
+                "",
+                "",
+                Collections.emptySet(),
+                false,
+                Collections.emptyList()));
+
+    Assert.assertEquals(TSStatusCode.USER_NOT_EXIST.getStatusCode(), status.getCode());
+  }
+
+  @Test
+  public void testRAccountUnlockRequiresExistingUser() throws Exception {
+    final IAuthorizer authorizer = mock(IAuthorizer.class);
+    when(authorizer.getUser("missing")).thenReturn(null);
+
+    final AuthorPlanExecutor executor = new AuthorPlanExecutor(authorizer);
+    final TSStatus status =
+        executor.executeRelationalAuthorNonQuery(
+            new AuthorRelationalPlan(
+                ConfigPhysicalPlanType.RAccountUnlock,
+                "missing",
+                "",
+                "",
+                "",
+                Collections.emptySet(),
+                false,
+                ""));
+
+    Assert.assertEquals(TSStatusCode.USER_NOT_EXIST.getStatusCode(), status.getCode());
+  }
+}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedureTest.java b/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();
+    }
   }
 }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java b/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(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java

Original file line number	Diff line number	Diff line change
`@@ -595,6 +595,7 @@ public void testIllegalPassword() throws Exception {`
`595`	`595`
`596`	`596`	`try {`
`597`	`597`	`statement.execute("alter pipe a2b modify source ('password'='fake')");`
	`598`	`+ fail();`
`598`	`599`	`} catch (final SQLException e) {`
`599`	`600`	`Assert.assertEquals("801: Failed to check password for pipe a2b.", e.getMessage());`
`600`	`601`	`}`