This is an automated email from the ASF dual-hosted git repository.
rong 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 db62fba21ba Subscription: implemented strict runtime permission check
for consumer group (#15400)
db62fba21ba is described below
commit db62fba21bace6d196f41c6e2f075b6eafe60613
Author: VGalaxies <[email protected]>
AuthorDate: Thu Apr 24 13:15:57 2025 +0800
Subscription: implemented strict runtime permission check for consumer
group (#15400)
---
.../it/local/IoTDBSubscriptionPermissionIT.java | 115 +++++++++++++++++++--
.../consumer/CreateConsumerProcedure.java | 4 +-
.../meta/consumer/ConsumerGroupMeta.java | 28 +++++
3 files changed, 140 insertions(+), 7 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionPermissionIT.java
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionPermissionIT.java
index 061366058d9..ce5a7df788a 100644
---
a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionPermissionIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionPermissionIT.java
@@ -37,8 +37,6 @@ import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.util.Arrays;
import java.util.Optional;
@@ -52,8 +50,6 @@ import static org.junit.Assert.fail;
@Category({LocalStandaloneIT.class})
public class IoTDBSubscriptionPermissionIT extends AbstractSubscriptionLocalIT
{
- private static final Logger LOGGER =
LoggerFactory.getLogger(IoTDBSubscriptionPermissionIT.class);
-
@Override
@Before
public void setUp() throws Exception {
@@ -131,6 +127,21 @@ public class IoTDBSubscriptionPermissionIT extends
AbstractSubscriptionLocalIT {
}
}
+ /**
+ * Tests runtime access control in the same consumer group.
+ *
+ * <p>In IoTDB subscriptions, all consumers in one group must use identical
credentials when
+ * subscribing to the same topic. This test creates a topic and three
consumers:
+ *
+ * <p>
+ *
+ * <ul>
+ * <li>consumer1 and consumer2 use "thulab:passwd".
+ * <li>consumer3 uses "hacker:qwerty123".
+ * </ul>
+ *
+ * <p>Since consumer3 uses different credentials, it should be rejected.
+ */
@Test
public void testRuntimeAccessControl() {
final String host = EnvFactory.getEnv().getIP();
@@ -223,12 +234,104 @@ public class IoTDBSubscriptionPermissionIT extends
AbstractSubscriptionLocalIT {
consumer1.open();
consumer1.subscribe(topicName);
-
consumer2.open();
consumer2.subscribe(topicName);
-
consumer3.open();
consumer3.subscribe(topicName);
+
+ fail();
+ } catch (final Exception e) {
+ }
+ }
+
+ /**
+ * Tests strict runtime access control in the same consumer group.
+ *
+ * <p>In IoTDB subscriptions, all consumers in one group must use identical
credentials. This test
+ * creates two consumers with "thulab:passwd" and one with
"hacker:qwerty123". Since the latter
+ * does not match the required credentials, it should be rejected.
+ */
+ @Test
+ public void testStrictRuntimeAccessControl() {
+ final String host = EnvFactory.getEnv().getIP();
+ final int port = Integer.parseInt(EnvFactory.getEnv().getPort());
+
+ // create user
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ EnvFactory.getEnv(),
+ Arrays.asList("create user `thulab` 'passwd'", "create user `hacker`
'qwerty123'"))) {
+ return;
+ }
+
+ final AtomicInteger rowCount = new AtomicInteger();
+ try (final SubscriptionTreePushConsumer consumer1 =
+ new SubscriptionTreePushConsumer.Builder()
+ .host(host)
+ .port(port)
+ .username("thulab")
+ .password("passwd")
+ .consumerId("thulab_consumer_1")
+ .consumerGroupId("thulab_consumer_group")
+ .ackStrategy(AckStrategy.AFTER_CONSUME)
+ .consumeListener(
+ message -> {
+ for (final SubscriptionSessionDataSet dataSet :
+ message.getSessionDataSetsHandler()) {
+ while (dataSet.hasNext()) {
+ dataSet.next();
+ rowCount.addAndGet(1);
+ }
+ }
+ return ConsumeResult.SUCCESS;
+ })
+ .buildPushConsumer();
+ final SubscriptionTreePushConsumer consumer2 =
+ new SubscriptionTreePushConsumer.Builder()
+ .host(host)
+ .port(port)
+ .username("thulab")
+ .password("passwd")
+ .consumerId("thulab_consumer_2")
+ .consumerGroupId("thulab_consumer_group")
+ .ackStrategy(AckStrategy.AFTER_CONSUME)
+ .consumeListener(
+ message -> {
+ for (final SubscriptionSessionDataSet dataSet :
+ message.getSessionDataSetsHandler()) {
+ while (dataSet.hasNext()) {
+ dataSet.next();
+ rowCount.addAndGet(1);
+ }
+ }
+ return ConsumeResult.SUCCESS;
+ })
+ .buildPushConsumer();
+ final SubscriptionTreePushConsumer consumer3 =
+ new SubscriptionTreePushConsumer.Builder()
+ .host(host)
+ .port(port)
+ .username("hacker")
+ .password("qwerty123")
+ .consumerId("hacker_consumer")
+ .consumerGroupId("thulab_consumer_group")
+ .ackStrategy(AckStrategy.AFTER_CONSUME)
+ .consumeListener(
+ message -> {
+ for (final SubscriptionSessionDataSet dataSet :
+ message.getSessionDataSetsHandler()) {
+ while (dataSet.hasNext()) {
+ dataSet.next();
+ rowCount.addAndGet(1);
+ }
+ }
+ return ConsumeResult.SUCCESS;
+ })
+ .buildPushConsumer()) {
+
+ consumer1.open();
+ consumer2.open();
+ consumer3.open();
+
fail();
} catch (final Exception e) {
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/CreateConsumerProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/CreateConsumerProcedure.java
index d9c10f5cbe7..1b961188d2f 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/CreateConsumerProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/CreateConsumerProcedure.java
@@ -67,11 +67,13 @@ public class CreateConsumerProcedure extends
AlterConsumerGroupProcedure {
createConsumerReq.getConsumerId(),
creationTime,
createConsumerReq.getConsumerAttributes());
- if (existingConsumerGroupMeta == null) {
+
+ if (Objects.isNull(existingConsumerGroupMeta)) {
updatedConsumerGroupMeta =
new ConsumerGroupMeta(
createConsumerReq.getConsumerGroupId(), creationTime,
newConsumerMeta);
} else {
+
existingConsumerGroupMeta.checkAuthorityBeforeJoinConsumerGroup(newConsumerMeta);
updatedConsumerGroupMeta = existingConsumerGroupMeta.deepCopy();
updatedConsumerGroupMeta.addConsumer(newConsumerMeta);
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
index 82871269220..13d010a636a 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
@@ -23,6 +23,8 @@ import
org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
import org.apache.tsfile.utils.PublicBAOS;
import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
@@ -38,6 +40,8 @@ import java.util.concurrent.ConcurrentHashMap;
public class ConsumerGroupMeta {
+ protected static final Logger LOGGER =
LoggerFactory.getLogger(ConsumerGroupMeta.class);
+
private String consumerGroupId;
private long creationTime;
private Map<String, Set<String>> topicNameToSubscribedConsumerIdSet;
@@ -102,6 +106,30 @@ public class ConsumerGroupMeta {
/////////////////////////////// consumer ///////////////////////////////
+ public void checkAuthorityBeforeJoinConsumerGroup(final ConsumerMeta
consumerMeta)
+ throws SubscriptionException {
+ if (isEmpty()) {
+ return;
+ }
+ final ConsumerMeta existedConsumerMeta =
consumerIdToConsumerMeta.values().iterator().next();
+ final boolean match =
+ Objects.equals(existedConsumerMeta.getUsername(),
consumerMeta.getUsername())
+ && Objects.equals(existedConsumerMeta.getPassword(),
consumerMeta.getPassword());
+ if (!match) {
+ final String exceptionMessage =
+ String.format(
+ "Failed to create consumer %s because inconsistent username &
password under the same consumer group, expected %s:%s, actual %s:%s",
+ consumerMeta.getConsumerId(),
+ existedConsumerMeta.getUsername(),
+ existedConsumerMeta.getPassword(),
+ consumerMeta.getUsername(),
+ consumerMeta.getPassword());
+ LOGGER.warn(exceptionMessage);
+ throw new SubscriptionException(exceptionMessage);
+ }
+ return;
+ }
+
public void addConsumer(final ConsumerMeta consumerMeta) {
consumerIdToConsumerMeta.put(consumerMeta.getConsumerId(), consumerMeta);
}
