Issue #540 - Better validation of application requests to prevent protocol errors

Signed-off-by: James Sutton <james.sutton@uk.ibm.com>

Author: jpwsutton

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java

@@ -1063,8 +1063,10 @@ public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext
 			throw new IllegalArgumentException();
 		}
 
-		// remove any message handlers for individual topics
+		// remove any message handlers for individual topics and validate topicFilter
 		for (int i = 0; i < topicFilters.length; ++i) {
+			// Check if the topic filter is valid before subscribing
+			MqttTopic.validate(topicFilters[i], true/* allow wildcards */);
 			this.comms.removeMessageListener(topicFilters[i]);
 		}
 
@@ -1077,8 +1079,7 @@ public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext
 				}
 				subs.append("topic=").append(topicFilters[i]).append(" qos=").append(qos[i]);
 
-				// Check if the topic filter is valid before subscribing
-				MqttTopic.validate(topicFilters[i], true/* allow wildcards */);
+				
 			}
 			// @TRACE 106=Subscribe topicFilter={0} userContext={1} callback={2}
 			log.fine(CLASS_NAME, methodName, "106", new Object[] { subs.toString(), userContext, callback });

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttAsyncClient.java

@@ -253,8 +253,13 @@ public class MqttAsyncClient implements MqttClientInterface, IMqttAsyncClient {
 												// second
 	private boolean reconnecting = false;
 	private static Object clientLock = new Object(); // Simple lock
-	private MqttSessionState mqttSession = new MqttSessionState(); // Variables that exist within the life of an MQTT session
-	private MqttConnectionState mqttConnection = new MqttConnectionState(); // Variables that exist within the life of an MQTT connection.
+
+	// Variables that exist within the life of an MQTT session
+	private MqttSessionState mqttSession = new MqttSessionState();
+
+	// Variables that exist within the life of an MQTT connection.
+	private MqttConnectionState mqttConnection = new MqttConnectionState(); 
+	
 	private ScheduledExecutorService executorService;
 	private MqttPingSender pingSender;
 
@@ -424,9 +429,9 @@ public MqttAsyncClient(String serverURI, String clientId) throws MqttException {
 	 * {@link MqttClientPersistence} interface. An implementer of this interface
 	 * that safely stores messages must be specified in order for delivery of
 	 * messages to be reliable. In addition
-	 * {@link MqttConnectionOptions#setCleanStart(boolean)} must be set to false.
-	 * In the event that only QoS 0 messages are sent or received or cleanStart is
-	 * set to true then a safe store is not needed.
+	 * {@link MqttConnectionOptions#setCleanStart(boolean)} must be set to false. In
+	 * the event that only QoS 0 messages are sent or received or cleanStart is set
+	 * to true then a safe store is not needed.
 	 * </p>
 	 * <p>
 	 * An implementation of file-based persistence is provided in class
@@ -527,9 +532,9 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence
 	 * {@link MqttClientPersistence} interface. An implementer of this interface
 	 * that safely stores messages must be specified in order for delivery of
 	 * messages to be reliable. In addition
-	 * {@link MqttConnectionOptions#setCleanStart(boolean)} must be set to false.
-	 * In the event that only QoS 0 messages are sent or received or cleanStart is
-	 * set to true then a safe store is not needed.
+	 * {@link MqttConnectionOptions#setCleanStart(boolean)} must be set to false. In
+	 * the event that only QoS 0 messages are sent or received or cleanStart is set
+	 * to true then a safe store is not needed.
 	 * </p>
 	 * <p>
 	 * An implementation of file-based persistence is provided in class
@@ -575,14 +580,13 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence
 			DataOutputStream dos = new DataOutputStream(baos);
 			MqttDataTypes.encodeUTF8(dos, clientId);
 			// Remove the two size bytes.
-			if(dos.size() - 2 > 65535) {
+			if (dos.size() - 2 > 65535) {
 				throw new IllegalArgumentException("ClientId longer than 65535 characters");
 			}
-			
+
 		} else {
 			clientId = "";
 		}
-		
 
 		MqttConnectionOptions.validateURI(serverURI);
 
@@ -608,7 +612,8 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence
 		log.fine(CLASS_NAME, methodName, "101", new Object[] { clientId, serverURI, persistence });
 
 		this.persistence.open(clientId);
-		this.comms = new ClientComms(this, this.persistence, this.pingSender, this.executorService, this.mqttSession, this.mqttConnection);
+		this.comms = new ClientComms(this, this.persistence, this.pingSender, this.executorService, this.mqttSession,
+				this.mqttConnection);
 		this.persistence.close();
 		this.topics = new Hashtable<String, MqttTopic>();
 
@@ -899,7 +904,7 @@ public IMqttToken connect(MqttConnectionOptions options, Object userContext, Mqt
 				userToken, userContext, callback, reconnecting, mqttSession, mqttConnection);
 		userToken.setActionCallback(connectActionListener);
 		userToken.setUserContext(this);
-		
+
 		this.mqttConnection.setSendReasonMessages(this.connOpts.isSendReasonMessages());
 
 		// If we are using the MqttCallbackExtended, set it on the
@@ -1134,7 +1139,7 @@ public String getCurrentServerURI() {
 	 *             if the topic contains a '+' or '#' wildcard character.
 	 */
 	protected MqttTopic getTopic(String topic) {
-		MqttTopicValidator.validate(topic, false/* wildcards NOT allowed */);
+		MqttTopicValidator.validate(topic, false/* wildcards NOT allowed */, true);
 
 		MqttTopic result = (MqttTopic) topics.get(topic);
 		if (result == null) {
@@ -1200,7 +1205,7 @@ public IMqttToken subscribe(String topicFilter, int qos) throws MqttException {
 		return this.subscribe(new MqttSubscription[] { new MqttSubscription(topicFilter, qos) }, null, null,
 				new MqttProperties());
 	}
-	
+
 	/*
 	 * (non-Javadoc)
 	 * 
@@ -1210,8 +1215,7 @@ public IMqttToken subscribe(String topicFilter, int qos) throws MqttException {
 	 */
 	@Override
 	public IMqttToken subscribe(MqttSubscription subscription) throws MqttException {
-		return this.subscribe(new MqttSubscription[] { subscription }, null, null,
-				new MqttProperties());
+		return this.subscribe(new MqttSubscription[] { subscription }, null, null, new MqttProperties());
 	}
 
 	/*
@@ -1240,9 +1244,13 @@ public IMqttToken subscribe(MqttSubscription[] subscriptions, Object userContext
 			MqttProperties subscriptionProperties) throws MqttException {
 		final String methodName = "subscribe";
 
-		// remove any message handlers for individual topics
+		// remove any message handlers for individual topics and validate Topics
 		for (int i = 0; i < subscriptions.length; ++i) {
 			this.comms.removeMessageListener(subscriptions[i].getTopic());
+			// Check if the topic filter is valid before subscribing
+			MqttTopicValidator.validate(subscriptions[i].getTopic(),
+					this.mqttConnection.isWildcardSubscriptionsAvailable(),
+					this.mqttConnection.isSharedSubscriptionsAvailable());
 		}
 
 		// Only Generate Log string if we are logging at FINE level
@@ -1253,9 +1261,6 @@ public IMqttToken subscribe(MqttSubscription[] subscriptions, Object userContext
 					subs.append(", ");
 				}
 				subs.append(subscriptions[i].toString());
-
-				// Check if the topic filter is valid before subscribing
-				MqttTopicValidator.validate(subscriptions[i].getTopic(), true/* allow wildcards */);
 			}
 			// @TRACE 106=Subscribe topicFilter={0} userContext={1} callback={2}
 			log.fine(CLASS_NAME, methodName, "106", new Object[] { subs.toString(), userContext, callback });
@@ -1365,14 +1370,15 @@ public IMqttToken subscribe(MqttSubscription[] subscriptions, Object userContext
 		int subId = subscriptionProperties.getSubscriptionIdentifiers().get(0);
 
 		// Automatic Subscription Identifier Assignment is enabled
-		if (connOpts.useSubscriptionIdentifiers()) {
+		if (connOpts.useSubscriptionIdentifiers() && this.mqttConnection.isSubscriptionIdentifiersAvailable()) {
 
 			// Application is overriding the subscription Identifier
 			if (subId != 0) {
 				// Check that we are not already using this ID, else throw Illegal Argument
 				// Exception
 				if (this.comms.doesSubscriptionIdentifierExist(subId)) {
-					throw new IllegalArgumentException(String.format("The Subscription Identifier %s already exists.", subId));
+					throw new IllegalArgumentException(
+							String.format("The Subscription Identifier %s already exists.", subId));
 				}
 
 			} else {
@@ -1459,7 +1465,7 @@ public IMqttToken unsubscribe(String[] topicFilters, Object userContext, MqttAct
 			// Although we already checked when subscribing, but invalid
 			// topic filter is meanless for unsubscribing, just prohibit it
 			// to reduce unnecessary control packet send to broker.
-			MqttTopicValidator.validate(topicFilters[i], true/* allow wildcards */);
+			MqttTopicValidator.validate(topicFilters[i], true/* allow wildcards */, this.mqttConnection.isSharedSubscriptionsAvailable());
 		}
 
 		// remove message handlers from the list for this client
@@ -1588,7 +1594,7 @@ public IMqttDeliveryToken publish(String topic, MqttMessage message, Object user
 		log.fine(CLASS_NAME, methodName, "111", new Object[] { topic, userContext, callback });
 
 		// Checks if a topic is valid when publishing a message.
-		MqttTopicValidator.validate(topic, false/* wildcards NOT allowed */);
+		MqttTopicValidator.validate(topic, false/* wildcards NOT allowed */, true);
 
 		MqttDeliveryToken token = new MqttDeliveryToken(getClientId());
 		token.setActionCallback(callback);

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttConnectionOptions.java

@@ -200,7 +200,7 @@ public void setWill(String topic, MqttMessage message) {
 		if (topic == null || message == null || message.getPayload() == null) {
 			throw new IllegalArgumentException();
 		}
-		MqttTopicValidator.validate(topic, false); // Wildcards are not allowed
+		MqttTopicValidator.validate(topic, false, true); // Wildcards are not allowed
 		this.willDestination = topic;
 		this.willMessage = message;
 		// Prevent any more changes to the will message

org.eclipse.paho.mqttv5.common/src/main/java/org/eclipse/paho/mqttv5/common/packet/MqttProperties.java

@@ -1234,7 +1234,7 @@ public String getResponseTopic() {
 	 */
 	public void setResponseTopic(String responseTopic) {
 		if(responseTopic != null) {
-			MqttTopicValidator.validate(responseTopic, false);
+			MqttTopicValidator.validate(responseTopic, false, true);
 		}
 		this.responseTopic = responseTopic;
 	}

org.eclipse.paho.mqttv5.common/src/main/java/org/eclipse/paho/mqttv5/common/util/MqttTopicValidator.java

@@ -38,7 +38,7 @@ public class MqttTopicValidator {
 	private static final int MIN_TOPIC_LEN = 1;
 	private static final int MAX_TOPIC_LEN = 65535;
 	private static final char NUL = '\u0000';
-	
+
 	/**
 	 * Validate the topic name or topic filter
 	 *
@@ -49,7 +49,8 @@ public class MqttTopicValidator {
 	 * @throws IllegalArgumentException
 	 *             if the topic is invalid
 	 */
-	public static void validate(String topicString, boolean wildcardAllowed) throws IllegalArgumentException {
+	public static void validate(String topicString, boolean wildcardAllowed, boolean sharedSubAllowed)
+			throws IllegalArgumentException {
 		int topicLen = 0;
 		try {
 			topicLen = topicString.getBytes("UTF-8").length;
@@ -103,12 +104,18 @@ public static void validate(String topicString, boolean wildcardAllowed) throws
 			return;
 		}
 
+		// Validate Shared Subscriptions
+		if (!sharedSubAllowed && topicString.startsWith("$share/")) {
+			throw new IllegalArgumentException("Shared Subscriptions are not allowed.");
+		}
+
 		// *******************************************************************************
 		// 2) This is a topic name string that MUST NOT contains any wildcard characters
 		// *******************************************************************************
 		if (Strings.containsAny(topicString, TOPIC_WILDCARDS)) {
 			throw new IllegalArgumentException("The topic name MUST NOT contain any wildcard characters (#+)");
 		}
+
 	}
 
 	private static void validateSingleLevelWildcard(String topicString) {
@@ -133,7 +140,7 @@ private static void validateSingleLevelWildcard(String topicString) {
 			}
 		}
 	}
-	
+
 	/**
 	 * Check the supplied topic name and filter match
 	 *
@@ -151,8 +158,8 @@ public static boolean isMatched(String topicFilter, String topicName) throws Ill
 		int topicLen = topicName.length();
 		int filterLen = topicFilter.length();
 
-		MqttTopicValidator.validate(topicFilter, true);
-		MqttTopicValidator.validate(topicName, false);
+		MqttTopicValidator.validate(topicFilter, true, true);
+		MqttTopicValidator.validate(topicName, false, true);
 
 		if (topicFilter.equals(topicName)) {
 			return true;
@@ -179,11 +186,11 @@ public static boolean isMatched(String topicFilter, String topicName) throws Ill
 			return true;
 		} else {
 			/*
-			 * https://github.com/eclipse/paho.mqtt.java/issues/418
-			 * Covers edge case to match sport/# to sport
+			 * https://github.com/eclipse/paho.mqtt.java/issues/418 Covers edge case to
+			 * match sport/# to sport
 			 */
-			if ((topicFilter.length() - topicName.length()) == 2 &&
-					topicFilter.substring(topicFilter.length() -2, topicFilter.length()).equals("/#")) {
+			if ((topicFilter.length() - topicName.length()) == 2
+					&& topicFilter.substring(topicFilter.length() - 2, topicFilter.length()).equals("/#")) {
 				String filterSub = topicFilter.substring(0, topicFilter.length() - 2);
 				if (filterSub.equals(topicName)) {
 					System.err.println("filterSub equals topicName: " + filterSub + " == " + topicName);
@@ -193,5 +200,5 @@ public static boolean isMatched(String topicFilter, String topicName) throws Ill
 		}
 		return false;
 	}
-	
+
 }

org.eclipse.paho.mqttv5.common/src/test/java/org/eclipse/paho/mqttv5/common/utils/MqttTopicTest.java

@@ -29,7 +29,7 @@ public void testValidTopicFilterWildcards() throws Exception {
 				"/+/sport/+/player1", "#", "/#", "sport/#", "sport/tennis/#" };
 
 		for (String topic : topics) {
-			MqttTopicValidator.validate(topic, true);
+			MqttTopicValidator.validate(topic, true, true);
 		}
 	}
 
@@ -57,32 +57,32 @@ public void testNonMatchedTopicFilterWildcards() throws Exception {
 
 	@Test(expected = IllegalArgumentException.class)
 	public void testInvalidTopicFilterWildcards1() throws Exception {
-		MqttTopicValidator.validate("sport/tennis#", true);
+		MqttTopicValidator.validate("sport/tennis#", true, true);
 	}
 
 	@Test(expected = IllegalArgumentException.class)
 	public void testInvalidTopicFilterWildcards2() throws Exception {
-		MqttTopicValidator.validate("sport/tennis/#/ranking", true);
+		MqttTopicValidator.validate("sport/tennis/#/ranking", true, true);
 	}
 
 	@Test(expected = IllegalArgumentException.class)
 	public void testInvalidTopicFilterWildcards3() throws Exception {
-		MqttTopicValidator.validate("sport+", true);
+		MqttTopicValidator.validate("sport+", true, true);
 	}
 
 	@Test(expected = IllegalArgumentException.class)
 	public void testInvalidTopicFilterWildcards4() throws Exception {
-		MqttTopicValidator.validate("sport/+aa", true);
+		MqttTopicValidator.validate("sport/+aa", true, true);
 	}
 
 	@Test(expected = IllegalArgumentException.class)
 	public void testInvalidTopicFilterWildcards5() throws Exception {
-		MqttTopicValidator.validate("sport/#/ball/+/aa", true);
+		MqttTopicValidator.validate("sport/#/ball/+/aa", true, true);
 	}
 
 	@Test
 	public void testValidG11NTopic() {
-		MqttTopicValidator.validate("$shared/葛渚噓/GVTDurTopic02/葛渚噓", true);
+		MqttTopicValidator.validate("$shared/葛渚噓/GVTDurTopic02/葛渚噓", true, true);
 	}
 
 }

org.eclipse.paho.mqttv5.common/src/test/java/org/eclipse/paho/mqttv5/common/utils/StringValidationTest.java

@@ -9,7 +9,7 @@ public class StringValidationTest {
 
 	@Test
 	public void testInvalidTopicFilterWildcards1() throws Exception {
-		MqttTopicValidator.validate(testString, true);
+		MqttTopicValidator.validate(testString, true, true);
 	}
 
 }