Add validation for topics before publishing.

Build/MQTTnet.nuspec

@@ -15,6 +15,7 @@
 * [Core] Fixed a memory leak when processing lots of messages (thanks to @tschanko)
 * [Core] Added more overloads for MQTT factory.
 * [Core] The client password is now hidden from the logs (replaced with **** if set).
+* [Client] Added validation of topics before publishing.
 * [Client] Added new MQTTv5 features to options builder.
 * [Client] Added uniform API across all supported MQTT versions (BREAKING CHANGE!)
 * [Client] The client will now avoid sending an ACK if an exception has been thrown in message handler (thanks to @ramonsmits).
@@ -22,13 +23,16 @@
 * [Client] Replaced all events with proper async compatible handlers (BREAKING CHANGE!).
 * [ManagedClient] Replaced all events with proper async compatible handlers (BREAKING CHANGE!).
 * [ManagedClient] The log ID is now propagated to the internal client (thanks to @vbBerni).
+* [ManagedClient] Added validation of topics before publishing.
+* [ManagedClient] The internal MQTT client is now closed properly (thanks to @vbBerni).
 * [Server] Added support for MQTTv5 clients. The server will still return _success_ for all cases at the moment even if more granular codes are available.
 * [Server] Fixed issues in QoS 2 handling which leads to message loss.
 * [Server] Replaced all events with proper async compatible handlers (BREAKING CHANGE!).
 * [Server] The used logger instance is now propagated to the WebSocket server adapter.
 * [Server] Added the flag "IsSecureConnection" which is set to true when the connection is encrypted.
 * [Server] Fixed wrong will message behavior when stopping server (thanks to @JohBa)
-* [MQTTnet Server] Added as first Alpha version.
+* [Server] Added validation of topics before publishing.
+* [MQTTnet Server] Added as first Alpha version of standalone cross platform MQTT server.
 
 * [Note] Due to MQTTv5 a lot of new classes were introduced. This required adding new namespaces as well. Most classes are backward compatible but new namespaces must be added.
     </releaseNotes>

README.md

@@ -54,13 +54,13 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
 
 ## MQTTnet Server
 
-_MQTTnet Server_ is a reference implementation of a MQTT server using this library. It has the following features.
+_MQTTnet Server_ is a standalone cross platform MQTT server (like mosquitto) basing on this library. It has the following features.
 * Running portable (no installation required)
-* Python scripting support for manipulating messages, validation of clients etc.
 * Runs und Windows, Linux, macOS, Raspberry Pi
+* Python scripting support for manipulating messages, validation of clients, building business logic etc.
 * Supports WebSocket and TCP (with and without TLS) connections
 * Provides a HTTP based API (including Swagger endpoint)
-* Extensive configuration
+* Extensive configuration parameters and customization supported
 
 ## Supported frameworks
 

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -127,6 +127,8 @@ namespace MQTTnet.Extensions.ManagedClient
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
+            MqttTopicValidator.ThrowIfInvalid(applicationMessage.ApplicationMessage.Topic);
+
             ManagedMqttApplicationMessage removedMessage = null;
             ApplicationMessageSkippedEventArgs applicationMessageSkippedEventArgs = null;
 

Source/MQTTnet/Client/MqttClient.cs

@@ -166,6 +166,8 @@ namespace MQTTnet.Client
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
+            MqttTopicValidator.ThrowIfInvalid(applicationMessage.Topic);
+
             ThrowIfNotConnected();
 
             var publishPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(applicationMessage);

Source/MQTTnet/Internal/AsyncAutoResetEvent.cs

@@ -74,8 +74,11 @@ namespace MQTTnet.Internal
             Task winner;
             if (timeout == Timeout.InfiniteTimeSpan)
             {
-                await tcs.Task.ConfigureAwait(false);
-                winner = tcs.Task;
+                using (cancellationToken.Register(() => { tcs.TrySetCanceled(); }))
+                {
+                    await tcs.Task.ConfigureAwait(false);
+                    winner = tcs.Task;
+                }               
             }
             else
             {
@@ -122,7 +125,7 @@ namespace MQTTnet.Internal
                 }
             }
 
-            toRelease?.SetResult(true);
+            toRelease?.TrySetResult(true);
         }
     }
 }

Source/MQTTnet/Protocol/MqttTopicValidator.cs

@@ -0,0 +1,25 @@
+using MQTTnet.Exceptions;
+
+namespace MQTTnet.Protocol
+{
+    public static class MqttTopicValidator
+    {
+        public static void ThrowIfInvalid(string topic)
+        {
+            if (string.IsNullOrEmpty(topic))
+            {
+                throw new MqttProtocolViolationException("Topic should not be empty.");
+            }
+
+            if (topic.Contains("+"))
+            {
+                throw new MqttProtocolViolationException("The character '+' is not allowed in topics.");
+            }
+
+            if (topic.Contains("#"))
+            {
+                throw new MqttProtocolViolationException("The character '#' is not allowed in topics.");
+            }
+        }
+    }
+}

Source/MQTTnet/Server/MqttServer.cs

@@ -7,6 +7,7 @@ using MQTTnet.Adapter;
 using MQTTnet.Client.Publishing;
 using MQTTnet.Client.Receiving;
 using MQTTnet.Diagnostics;
+using MQTTnet.Protocol;
 using MQTTnet.Server.Status;
 
 namespace MQTTnet.Server
@@ -101,6 +102,8 @@ namespace MQTTnet.Server
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
+            MqttTopicValidator.ThrowIfInvalid(applicationMessage.Topic);
+
             if (_cancellationTokenSource == null) throw new InvalidOperationException("The server is not started.");
 
             _clientSessionsManager.DispatchApplicationMessage(applicationMessage, null);

Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs

@@ -123,13 +123,12 @@ namespace MQTTnet.Tests
             }
             catch (OperationCanceledException ex)
             {
-                Assert.AreEqual(cts.Token, ex.CancellationToken);
             }
 
             // Now set the event and verify that a future waiter gets the signal immediately.
             _aare.Set();
             waitTask = _aare.WaitOneAsync();
-            Assert.AreEqual(TaskStatus.RanToCompletion, waitTask.Status);
+            Assert.AreEqual(TaskStatus.WaitingForActivation, waitTask.Status);
         }
 
         [TestMethod]

Tests/MQTTnet.Core.Tests/MqttTopicValidator_Tests.cs

@@ -0,0 +1,37 @@
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Exceptions;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Tests
+{
+    [TestClass]
+    public class MqttTopicValidator_Tests
+    {
+        [TestMethod]
+        public void Valid_Topic()
+        {
+            MqttTopicValidator.ThrowIfInvalid("/a/b/c");
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(MqttProtocolViolationException))]
+        public void Invalid_Topic_Plus()
+        {
+            MqttTopicValidator.ThrowIfInvalid("/a/+/c");
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(MqttProtocolViolationException))]
+        public void Invalid_Topic_Hash()
+        {
+            MqttTopicValidator.ThrowIfInvalid("/a/#/c");
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(MqttProtocolViolationException))]
+        public void Invalid_Topic_Empty()
+        {
+            MqttTopicValidator.ThrowIfInvalid(string.Empty);
+        }
+    }
+}