Skip to content

Fix account unlock propagation for pipe password checks #17814

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Fix account unlock propagation for pipe password checks #17814

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions ...t/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}
Expand Down
2 changes: 2 additions & 0 deletions ...gnode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down
1 change: 1 addition & 0 deletions ...e/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,6 +333,7 @@ public enum ConfigPhysicalPlanType {
RDropUserV2((short) 2103),
RenameUser((short) 2104),
RRenameUser((short) 2105),
AccountUnlock((short) 2106),

EnableSeparationOfAdminPowers((short) 2200),

Expand Down
12 changes: 12 additions & 0 deletions ...rc/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanVisitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -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);
}
Expand Down
6 changes: 5 additions & 1 deletion ...ore/confignode/src/main/java/org/apache/iotdb/confignode/persistence/auth/AuthorInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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()[
Expand Down
4 changes: 4 additions & 0 deletions ...ignode/src/main/java/org/apache/iotdb/confignode/persistence/auth/AuthorPlanExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down
1 change: 1 addition & 0 deletions ...de/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -488,6 +488,7 @@ public TSStatus executeNonQueryPlan(ConfigPhysicalPlan physicalPlan)
case RevokeRoleFromUserDep:
case UpdateUserDep:
case RenameUser:
case AccountUnlock:
case RCreateRole:
case RCreateUser:
case RDropUser:
Expand Down
6 changes: 6 additions & 0 deletions ...src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand Down
56 changes: 56 additions & 0 deletions ...test/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedureTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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();
}
}
}
35 changes: 11 additions & 24 deletions iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down Expand Up @@ -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(
Expand Down
14 changes: 14 additions & 0 deletions ...rc/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@
import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException;
import org.apache.iotdb.db.audit.DNAuditLogger;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.auth.LoginLockManager;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
import org.apache.iotdb.db.consensus.SchemaRegionConsensusImpl;
Expand Down Expand Up @@ -2451,12 +2452,25 @@ private void sampleDiskLoad(TLoadSample loadSample) {

@Override
public TSStatus invalidatePermissionCache(TInvalidatePermissionCacheReq req) {
if (req.isSetNeedDisconnect() && req.isNeedDisconnect()) {
return unlockAccountAndInvalidateCache(req);
}
if (AuthorityChecker.invalidateCache(req.getUsername(), req.getRoleName())) {
return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
}
return RpcUtils.getStatus(TSStatusCode.CLEAR_PERMISSION_CACHE_ERROR);
}

private TSStatus unlockAccountAndInvalidateCache(TInvalidatePermissionCacheReq req) {
// For account-unlock broadcasts, roleName carries the optional login address.
AuthorityChecker.getUserId(req.getUsername())
.ifPresent(userId -> LoginLockManager.getInstance().unlock(userId, req.getRoleName()));
if (AuthorityChecker.invalidateCache(req.getUsername(), null)) {
return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
}
return RpcUtils.getStatus(TSStatusCode.CLEAR_PERMISSION_CACHE_ERROR);
}

@Override
public TSStatus enableSeparationOfAdminPower() throws TException {
return null;
Expand Down
Loading