diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java index 1752c8acf767..790e022ded36 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java @@ -3485,7 +3485,7 @@ public TPipeSubscribeResp pipeSubscribe(final TPipeSubscribeReq req) { return getNotLoggedInPipeSubscribeResp(); } - return SubscriptionAgent.receiver().handle(req); + return SubscriptionAgent.receiver().handle(req, clientSession.getUsername()); } finally { SESSION_MANAGER.updateIdleTime(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java index 4b26a5668374..192a07dc83a1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java @@ -78,6 +78,16 @@ public class SubscriptionReceiverAgent { } public TPipeSubscribeResp handle(final TPipeSubscribeReq req) { + return handle(req, null); + } + + public TPipeSubscribeResp handle(final TPipeSubscribeReq req, final String username) { + if (username == null) { + return new TPipeSubscribeResp( + RpcUtils.getStatus(TSStatusCode.NO_PERMISSION), + PipeSubscribeResponseVersion.VERSION_1.getVersion(), + PipeSubscribeResponseType.ACK.getType()); + } if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { return SUBSCRIPTION_NOT_ENABLED_ERROR_RESP; } @@ -85,6 +95,7 @@ public TPipeSubscribeResp handle(final TPipeSubscribeReq req) { final byte reqVersion = req.getVersion(); if (RECEIVER_CONSTRUCTORS.containsKey(reqVersion)) { final SubscriptionReceiver receiver = getReceiver(reqVersion); + receiver.setAuthenticatedUsername(username); activeReceivers.add(receiver); receiver.handleTimeout(); return receiver.handle(req); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java index 3f69633f499f..256c87e521dc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java @@ -20,9 +20,15 @@ package org.apache.iotdb.db.subscription.agent; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.auth.entity.PrivilegeType; +import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; +import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern; +import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern; +import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; import org.apache.iotdb.commons.subscription.meta.topic.TopicMetaKeeper; +import org.apache.iotdb.db.auth.AuthorityChecker; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.mpp.rpc.thrift.TPushTopicMetaRespExceptionMessage; @@ -369,6 +375,105 @@ public TSStatus checkTopicOwners( return RpcUtils.SUCCESS_STATUS; } + /** + * Check that the authenticated session can read all data covered by the requested topics. + * ConsumerConfig is client-controlled and therefore must not be used as the authorization + * identity. + */ + public TSStatus checkTopicReadPermissions( + final String username, + final ConsumerConfig consumerConfig, + final Iterable topicNames) { + if (Objects.isNull(username)) { + return RpcUtils.getStatus(TSStatusCode.NO_PERMISSION); + } + + acquireReadLock(); + try { + for (final String topicName : topicNames) { + final TopicMeta topicMeta = + topicMetaKeeper.getTopicMeta(topicName, isTableModel(consumerConfig)); + if (Objects.isNull(topicMeta)) { + continue; + } + + final TSStatus status = + topicMeta.getConfig().isTableTopic() + ? checkTableTopicReadPermission(username, topicMeta) + : checkTreeTopicReadPermission(username, topicMeta); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + return status; + } + } + return RpcUtils.SUCCESS_STATUS; + } finally { + releaseReadLock(); + } + } + + private TSStatus checkTreeTopicReadPermission(final String username, final TopicMeta topicMeta) { + final TopicConfig topicConfig = topicMeta.getConfig(); + final TreePattern treePattern = + topicConfig.getAttribute().containsKey(TopicConstant.PATTERN_KEY) + ? new PrefixTreePattern(topicConfig.getAttribute().get(TopicConstant.PATTERN_KEY)) + : new IoTDBTreePattern( + topicConfig.getStringOrDefault( + TopicConstant.PATH_KEY, TopicConstant.PATH_DEFAULT_VALUE)); + for (final PartialPath path : treePattern.getBaseInclusionPaths()) { + if (!AuthorityChecker.checkFullPathOrPatternPermission( + username, path, PrivilegeType.READ_DATA)) { + return AuthorityChecker.getTSStatus(false, path, PrivilegeType.READ_DATA); + } + } + return RpcUtils.SUCCESS_STATUS; + } + + private TSStatus checkTableTopicReadPermission(final String username, final TopicMeta topicMeta) { + if (AuthorityChecker.SUPER_USER.equals(username)) { + return RpcUtils.SUCCESS_STATUS; + } + final TopicConfig topicConfig = topicMeta.getConfig(); + final String database = + topicConfig.getStringOrDefault( + TopicConstant.DATABASE_KEY, TopicConstant.DATABASE_DEFAULT_VALUE); + final String table = + topicConfig.getStringOrDefault(TopicConstant.TABLE_KEY, TopicConstant.TABLE_DEFAULT_VALUE); + + // A database-level SELECT grant covers all tables in one database. For a topic whose + // database/table is a regular expression, only an any-scope SELECT grant is broad enough to + // cover every object matched by the topic. + final boolean databasePattern = isRegexPattern(database); + final boolean tablePattern = isRegexPattern(table); + final boolean allowed = + (databasePattern + ? AuthorityChecker.checkDBPermission( + username, AuthorityChecker.ANY_SCOPE, PrivilegeType.SELECT) + : AuthorityChecker.checkDBPermission(username, database, PrivilegeType.SELECT) + || (!tablePattern + && AuthorityChecker.checkTablePermission( + username, database, table, PrivilegeType.SELECT))); + return allowed + ? RpcUtils.SUCCESS_STATUS + : AuthorityChecker.getTSStatus(false, PrivilegeType.SELECT, database, table); + } + + private static boolean isRegexPattern(final String value) { + return value.indexOf('.') >= 0 + || value.indexOf('*') >= 0 + || value.indexOf('+') >= 0 + || value.indexOf('?') >= 0 + || value.indexOf('[') >= 0 + || value.indexOf(']') >= 0 + || value.indexOf('(') >= 0 + || value.indexOf(')') >= 0 + || value.indexOf('{') >= 0 + || value.indexOf('}') >= 0 + || value.indexOf('|') >= 0 + || value.indexOf('^') >= 0 + || value.indexOf('$') >= 0 + || value.indexOf('\\') >= 0; + } + /** * Apply owner lease renewals pushed by ConfigNode via the dedicated subscription owner heartbeat. * The pushed remaining duration is converted to a DataNode-local expire time on the local clock, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiver.java index 36e3c9b74f5e..cc7b57eee81c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiver.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiver.java @@ -27,6 +27,8 @@ public interface SubscriptionReceiver { TPipeSubscribeResp handle(TPipeSubscribeReq req); + void setAuthenticatedUsername(final String username); + PipeSubscribeRequestVersion getVersion(); void handleExit(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java index 268bb8df1191..4ca93fb5cb8d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java @@ -123,6 +123,7 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver { private final ThreadLocal consumerConfigThreadLocal = new ThreadLocal<>(); private final ThreadLocal pollTimerThreadLocal = new ThreadLocal<>(); + private volatile String authenticatedUsername; private volatile ConsumerConfig sharedConsumerConfig; private volatile boolean consumerInvalidated; private volatile long lastActivityTimeMs = System.currentTimeMillis(); @@ -180,6 +181,11 @@ public PipeSubscribeRequestVersion getVersion() { return PipeSubscribeRequestVersion.VERSION_1; } + @Override + public void setAuthenticatedUsername(final String username) { + authenticatedUsername = username; + } + @Override public void handleExit() { final ConsumerConfig consumerConfig = consumerConfigThreadLocal.get(); @@ -197,6 +203,7 @@ public void handleExit() { consumerConfigThreadLocal.remove(); } clearSharedConsumerState(); + authenticatedUsername = null; } @Override @@ -369,6 +376,13 @@ private TPipeSubscribeResp handleSubscriptionHeartbeatInternal(final Subscriptio return PipeSubscribeHeartbeatResp.toTPipeSubscribeResp(ownerStatus); } + final TSStatus readPermissionStatus = + SubscriptionAgent.topic() + .checkTopicReadPermissions(authenticatedUsername, consumerConfig, subscribedTopicNames); + if (readPermissionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + return PipeSubscribeHeartbeatResp.toTPipeSubscribeResp(readPermissionStatus); + } + LOGGER.info(DataNodeMiscMessages.SUBSCRIPTION_CONSUMER_HEARTBEAT_SUCCESS, consumerConfig); // fetch subscribed topics @@ -453,6 +467,12 @@ private TPipeSubscribeResp handlePipeSubscribeSubscribeInternal( if (ownerStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { return PipeSubscribeSubscribeResp.toTPipeSubscribeResp(ownerStatus); } + final TSStatus readPermissionStatus = + SubscriptionAgent.topic() + .checkTopicReadPermissions(authenticatedUsername, consumerConfig, topicNames); + if (readPermissionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + return PipeSubscribeSubscribeResp.toTPipeSubscribeResp(readPermissionStatus); + } subscribe(consumerConfig, topicNames); LOGGER.info( @@ -564,6 +584,14 @@ private TPipeSubscribeResp handlePipeSubscribePollInternal(final PipeSubscribePo if (ownerStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { return PipeSubscribePollResp.toTPipeSubscribeResp(ownerStatus, Collections.emptyList()); } + final TSStatus readPermissionStatus = + SubscriptionAgent.topic() + .checkTopicReadPermissions( + authenticatedUsername, consumerConfig, topicNamesToCheck); + if (readPermissionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + return PipeSubscribePollResp.toTPipeSubscribeResp( + readPermissionStatus, Collections.emptyList()); + } events = handlePipeSubscribePollRequest( consumerConfig, @@ -581,6 +609,18 @@ private TPipeSubscribeResp handlePipeSubscribePollInternal(final PipeSubscribePo return PipeSubscribePollResp.toTPipeSubscribeResp( tsFileOwnerStatus, Collections.emptyList()); } + final String tsFileTopicName = + ((PollFilePayload) request.getPayload()).getCommitContext().getTopicName(); + final TSStatus tsFileReadPermissionStatus = + SubscriptionAgent.topic() + .checkTopicReadPermissions( + authenticatedUsername, + consumerConfig, + Collections.singleton(tsFileTopicName)); + if (tsFileReadPermissionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + return PipeSubscribePollResp.toTPipeSubscribeResp( + tsFileReadPermissionStatus, Collections.emptyList()); + } events = handlePipeSubscribePollTsFileRequest( consumerConfig, (PollFilePayload) request.getPayload()); @@ -597,6 +637,19 @@ private TPipeSubscribeResp handlePipeSubscribePollInternal(final PipeSubscribePo return PipeSubscribePollResp.toTPipeSubscribeResp( tabletsOwnerStatus, Collections.emptyList()); } + final String tabletsTopicName = + ((PollTabletsPayload) request.getPayload()).getCommitContext().getTopicName(); + final TSStatus tabletsReadPermissionStatus = + SubscriptionAgent.topic() + .checkTopicReadPermissions( + authenticatedUsername, + consumerConfig, + Collections.singleton(tabletsTopicName)); + if (tabletsReadPermissionStatus.getCode() + != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + return PipeSubscribePollResp.toTPipeSubscribeResp( + tabletsReadPermissionStatus, Collections.emptyList()); + } events = handlePipeSubscribePollTabletsRequest( consumerConfig, (PollTabletsPayload) request.getPayload());