This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new bc5b6a2d162 Fix subscription topic authorization bypass (#18418)
bc5b6a2d162 is described below
commit bc5b6a2d162fa1f38c0553fe177430629bf55262
Author: Caideyipi <[email protected]>
AuthorDate: Mon Aug 10 09:47:35 2026 +0800
Fix subscription topic authorization bypass (#18418)
---
.../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());
