You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/08/07 03:01:14 UTC
(iotdb) branch fix/subscription-rbac-transfer-slice-dos created (now ed94e550994)
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a change to branch fix/subscription-rbac-transfer-slice-dos
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at ed94e550994 Fix subscription topic authorization bypass
This branch includes the following new commits:
new ed94e550994 Fix subscription topic authorization bypass
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: Fix subscription topic authorization bypass
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch fix/subscription-rbac-transfer-slice-dos
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ed94e550994992a949b5554a4a32d183f4435efe
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Fri Aug 7 11:00:13 2026 +0800
Fix subscription topic authorization bypass
---
.../protocol/thrift/impl/ClientRPCServiceImpl.java | 2 +-
.../agent/SubscriptionReceiverAgent.java | 11 +++
.../subscription/agent/SubscriptionTopicAgent.java | 105 +++++++++++++++++++++
.../receiver/SubscriptionReceiver.java | 2 +
.../receiver/SubscriptionReceiverV1.java | 53 +++++++++++
5 files changed, 172 insertions(+), 1 deletion(-)
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 1752c8acf76..790e022ded3 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 class ClientRPCServiceImpl implements IClientRPCServiceWithHandler {
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 4b26a566837..192a07dc83a 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 class SubscriptionReceiverAgent {
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 3f69633f499..256c87e521d 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 class SubscriptionTopicAgent {
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<String> 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 36e3c9b74f5..cc7b57eee81 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 268bb8df119..4ca93fb5cb8 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<ConsumerConfig> consumerConfigThreadLocal = new ThreadLocal<>();
private final ThreadLocal<PollTimer> 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 class SubscriptionReceiverV1 implements SubscriptionReceiver {
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 class SubscriptionReceiverV1 implements SubscriptionReceiver {
consumerConfigThreadLocal.remove();
}
clearSharedConsumerState();
+ authenticatedUsername = null;
}
@Override
@@ -369,6 +376,13 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver {
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 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver {
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 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver {
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 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver {
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 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver {
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());