From 46234e79efb9cecf2b05beb16881d817bbe46b6c Mon Sep 17 00:00:00 2001 From: "Christopher L. Shannon" Date: Thu, 3 Sep 2026 16:44:26 -0400 Subject: [PATCH] Validation subscription key length in KahaDB Prevent subscription from exceeding max length of 65535 --- .../activemq/store/kahadb/KahaDBStore.java | 7 +++- .../store/kahadb/KahaDBStoreTest.java | 32 +++++++++++++++++++ .../DurableSubscriptionTestSupport.java | 6 ++-- .../KahaDBDurableSubscriptionTest.java | 18 +++++++++++ 4 files changed, 59 insertions(+), 4 deletions(-) diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java index 2f809d403e7..dbf52cd905c 100644 --- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java +++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java @@ -103,6 +103,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, public static final String PROPERTY_ASYNC_EXECUTOR_MAX_THREADS = "org.apache.activemq.store.kahadb.ASYNC_EXECUTOR_MAX_THREADS"; private static final int asyncExecutorMaxThreads = Integer.parseInt(System.getProperty( PROPERTY_ASYNC_EXECUTOR_MAX_THREADS, "1"), 10); + static final int MAX_SUBSCRIPTION_KEY_LENGTH = 65535; protected ExecutorService queueExecutor; protected ExecutorService topicExecutor; @@ -1403,7 +1404,11 @@ public MessageStoreSubscriptionStatistics getMessageStoreSubStatistics() { } String subscriptionKey(String clientId, String subscriptionName) { - return clientId + ":" + subscriptionName; + String key = clientId + ":" + subscriptionName; + if (key.length() > MAX_SUBSCRIPTION_KEY_LENGTH) { + throw new IllegalArgumentException("Subscription key exceeds max length " + MAX_SUBSCRIPTION_KEY_LENGTH); + } + return key; } @Override diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBStoreTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBStoreTest.java index edfb006c48b..f94c4b50389 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBStoreTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBStoreTest.java @@ -16,6 +16,7 @@ */ package org.apache.activemq.store.kahadb; +import java.io.IOException; import java.util.ArrayList; import java.util.Vector; import java.util.concurrent.ExecutorService; @@ -24,16 +25,20 @@ import org.apache.activemq.command.ActiveMQDestination; import org.apache.activemq.command.ActiveMQMessage; import org.apache.activemq.command.ActiveMQQueue; +import org.apache.activemq.command.ActiveMQTopic; import org.apache.activemq.command.Message; import org.apache.activemq.command.MessageAck; import org.apache.activemq.command.MessageId; import org.apache.activemq.command.ProducerId; +import org.apache.activemq.command.SubscriptionInfo; +import org.apache.activemq.store.TopicMessageStore; import org.junit.After; import org.junit.Before; import org.junit.Test; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; public class KahaDBStoreTest { @@ -139,4 +144,31 @@ public void testTrackAndForgetRecoveredAcks() throws Exception { underTest.forgetRecoveredAcks(new ArrayList<>(acks.subList(5, 10)), false); assertTrue(underTest.ackedAndPreparedMap.isEmpty()); } + + @Test + public void testSubscriptionKeyValidation() { + assertEquals("client1:sub1", store.subscriptionKey("client1", "sub1")); + Throwable error = assertThrows(IllegalArgumentException.class, + () -> store.subscriptionKey("client1", "a".repeat(KahaDBStore.MAX_SUBSCRIPTION_KEY_LENGTH))); + assertTrue(error.getMessage().contains("Subscription key exceeds max length")); + error = assertThrows(IllegalArgumentException.class, + () -> store.subscriptionKey("a".repeat(KahaDBStore.MAX_SUBSCRIPTION_KEY_LENGTH), "subname")); + assertTrue(error.getMessage().contains("Subscription key exceeds max length")); + } + + @Test + public void testSubscriptionCommandValidation() throws IOException { + TopicMessageStore topicStore = store.createTopicMessageStore(new ActiveMQTopic("test")); + + // check addSubscription, which creates KahaSubscriptionCommand, validates + Throwable error = assertThrows(IllegalArgumentException.class, + () -> topicStore.addSubscription(new SubscriptionInfo("client1", + "a".repeat(KahaDBStore.MAX_SUBSCRIPTION_KEY_LENGTH)), false)); + assertTrue(error.getMessage().contains("Subscription key exceeds max length")); + + // check deleteSubscription, which also creates KahaSubscriptionCommand, validates + error = assertThrows(IllegalArgumentException.class, () -> topicStore.deleteSubscription( + "client1", "a".repeat(KahaDBStore.MAX_SUBSCRIPTION_KEY_LENGTH))); + assertTrue(error.getMessage().contains("Subscription key exceeds max length")); + } } diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionTestSupport.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionTestSupport.java index 1913c3bda56..0ceb0850821 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionTestSupport.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionTestSupport.java @@ -40,9 +40,9 @@ public abstract class DurableSubscriptionTestSupport extends TestSupport { protected Connection connection; - private Session session; - private TopicSubscriber consumer; - private MessageProducer producer; + protected Session session; + protected TopicSubscriber consumer; + protected MessageProducer producer; protected BrokerService broker; @Override diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/KahaDBDurableSubscriptionTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/KahaDBDurableSubscriptionTest.java index 436c1cd4e09..559a48aaba7 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/KahaDBDurableSubscriptionTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/KahaDBDurableSubscriptionTest.java @@ -16,6 +16,11 @@ */ package org.apache.activemq.usecases; +import static org.junit.Assert.assertThrows; + +import jakarta.jms.DeliveryMode; +import jakarta.jms.JMSException; +import jakarta.jms.Topic; import java.io.IOException; import org.apache.activemq.store.PersistenceAdapter; import org.junit.experimental.categories.Category; @@ -27,4 +32,17 @@ public class KahaDBDurableSubscriptionTest extends DurableSubscriptionTestSuppor protected PersistenceAdapter createPersistenceAdapter() throws IOException { return null; // use default } + + public void testDurableSubscriptionKeyValidation() throws Exception { + connection.start(); + session = connection.createSession(false, jakarta.jms.Session.AUTO_ACKNOWLEDGE); + + Topic topic = session.createTopic("TestTopic"); + // This should throw an exception back to the client because the sub key + // value is too large and validation should catch it + assertThrows(JMSException.class, () -> session.createDurableSubscriber(topic, "s".repeat(65535))); + + // Ensure we can restart the broker without error + restartBroker(); + } }