Validate MQTT 5 SUBSCRIBE identifiers (#2262)

Observe packet task failures to prevent unobserved overflow exceptions

Server dispatch and session enqueue operations create packet bus items without
waiting for delivery. When the pending-message queue overflows, Fail faults
the item's completion task, which can later raise UnobservedTaskException
even if the application handles the dropped-message notification.

Observe the completion task's exception inside MqttPacketBusItem.Fail. This
keeps the task faulted and preserves the original exception for existing and
future WaitAsync callers, including DeliverApplicationMessageAsync.

Add a garbage-collection regression test for failures without a waiter and
verify that requesting the task before or after failure preserves the original
overflow exception. Re-enable unobserved-exception tracking in the existing
overflow integration tests for both drop strategies.

Before the fix, the focused run failed the new unobserved-exception test and
both enqueue overflow tests. After the fix, all nine focused cases and all
489 tests in the .NET 10 Release test suite pass.

Fixes https://github.com/dotnet/MQTTnet/issues/2271

README.md

@@ -12,9 +12,9 @@
 
 # MQTTnet
 
-MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (
+MQTTnet is a high performance .NET library for MQTT based communication.
-broker) and supports the MQTT protocol up to version 5. It is compatible with mostly any supported .NET Framework
+It provides a MQTT client and a MQTT server (broker) and supports the MQTT protocol up to version 5.
-version and CPU architecture.
+It is compatible with mostly any supported .NET Framework version and CPU architecture.
 
 ## Features
 

Source/MQTTnet.Tests/Formatter/MqttV5PublishPropertyValidation_Tests.cs

@@ -0,0 +1,40 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using MQTTnet.Exceptions;
+using MQTTnet.Formatter;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Tests.Formatter;
+
+[TestClass]
+public sealed class MqttV5PublishPropertyValidation_Tests
+{
+    [TestMethod]
+    [DataRow("0100")]
+    [DataRow("0200000000")]
+    [DataRow("030000")]
+    [DataRow("08000161")]
+    [DataRow("090000")]
+    [DataRow("230001")]
+    public void Publish_Rejects_Duplicate_Singleton_Property(string propertyHex)
+    {
+        var property = Convert.FromHexString(propertyHex);
+        byte[] packet = [0x30, (byte)(4 + 2 * property.Length), 0, 1, (byte)'a', (byte)(2 * property.Length), .. property, .. property];
+
+        Assert.ThrowsExactly<MqttProtocolViolationException>(() => MqttPacketSerializationHelper.DecodePacket(packet, MqttProtocolVersion.V500));
+    }
+
+    [TestMethod]
+    public void Publish_Allows_Repeated_Subscription_Identifiers_And_User_Properties()
+    {
+        byte[] packet = [0x30, 18, 0, 1, (byte)'a', 14, 0x0B, 1, 0x26, 0, 0, 0, 0, 0x0B, 1, 0x26, 0, 0, 0, 0];
+        var decoded = (MqttPublishPacket)MqttPacketSerializationHelper.DecodePacket(packet, MqttProtocolVersion.V500);
+
+        CollectionAssert.AreEqual(new uint[] { 1, 1 }, decoded.SubscriptionIdentifiers);
+        Assert.HasCount(2, decoded.UserProperties);
+    }
+
+
+}

Source/MQTTnet.Tests/Formatter/MqttV5SubscribeIdentifier_Tests.cs

@@ -0,0 +1,62 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using MQTTnet.Adapter;
+using MQTTnet.Exceptions;
+using MQTTnet.Formatter;
+using MQTTnet.Formatter.V5;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Tests.Formatter;
+
+[TestClass]
+public sealed class MqttV5SubscribeIdentifier_Tests
+{
+    [TestMethod]
+    [DataRow("0B00")]
+    [DataRow("0B010B01")]
+    [DataRow("0B010B02")]
+    [DataRow("0B0126000000000B02")]
+    public void Reject_Invalid_Subscription_Identifier(string propertiesHex)
+    {
+        var properties = Convert.FromHexString(propertiesHex);
+        byte[] packet = [0x82, (byte)(7 + properties.Length), 0, 1, (byte)properties.Length, .. properties, 0, 1, (byte)'a', 0];
+
+        Assert.ThrowsExactly<MqttProtocolViolationException>(() => MqttPacketSerializationHelper.DecodePacket(packet, MqttProtocolVersion.V500));
+    }
+
+    [TestMethod]
+    [DataRow("", 0U)]
+    [DataRow("0B01", 1U)]
+    [DataRow("0BFFFFFF7F", 268435455U)]
+    [DataRow("26000000000B012600000000", 1U)]
+    public void Accept_Valid_Subscription_Identifier(string propertiesHex, uint expected)
+    {
+        var properties = Convert.FromHexString(propertiesHex);
+        byte[] packet = [0x82, (byte)(7 + properties.Length), 0, 1, (byte)properties.Length, .. properties, 0, 1, (byte)'a', 0];
+        var decoded = (MqttSubscribePacket)MqttPacketSerializationHelper.DecodePacket(packet, MqttProtocolVersion.V500);
+
+        Assert.AreEqual(expected, decoded.SubscriptionIdentifier);
+    }
+
+    [TestMethod]
+    public void Accept_Same_Identifier_In_Separate_Subscribe_Packets()
+    {
+        byte[] packet = [0x82, 9, 0, 1, 2, 0x0B, 1, 0, 1, (byte)'a', 0];
+        var decoder = new MqttV5PacketDecoder();
+        var received = new ReceivedMqttPacket(packet[0], new ArraySegment<byte>(packet, 2, packet.Length - 2), packet.Length);
+
+        Assert.AreEqual(1U, ((MqttSubscribePacket)decoder.Decode(received)).SubscriptionIdentifier);
+        Assert.AreEqual(1U, ((MqttSubscribePacket)decoder.Decode(received)).SubscriptionIdentifier);
+    }
+
+    [TestMethod]
+    public void Accept_Repeated_Subscription_Identifiers_In_Publish()
+    {
+        byte[] packet = [0x30, 8, 0, 1, (byte)'a', 4, 0x0B, 1, 0x0B, 1];
+        var decoded = (MqttPublishPacket)MqttPacketSerializationHelper.DecodePacket(packet, MqttProtocolVersion.V500);
+
+        CollectionAssert.AreEqual(new uint[] { 1, 1 }, decoded.SubscriptionIdentifiers);
+    }
+}

Source/MQTTnet.Tests/Internal/MqttPacketBusItem_Tests.cs

@@ -2,8 +2,11 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+using System.Runtime.CompilerServices;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
+using MQTTnet.Server;
+using MQTTnet.Server.Exceptions;
 
 namespace MQTTnet.Tests.Internal;
 
@@ -11,6 +14,57 @@ namespace MQTTnet.Tests.Internal;
 [TestClass]
 public sealed class MqttPacketBusItem_Tests
 {
+    [TestMethod]
+    public void Fail_Without_Waiter_Does_Not_Raise_Unobserved_Exception()
+    {
+        var exception = new MqttPendingMessagesOverflowException("test", MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage);
+        var unobservedExceptions = 0;
+
+        void OnUnobservedTaskException(object sender, UnobservedTaskExceptionEventArgs args)
+        {
+            if (args.Exception.Flatten().InnerExceptions.Contains(exception))
+            {
+                Interlocked.Increment(ref unobservedExceptions);
+                args.SetObserved();
+            }
+        }
+
+        TaskScheduler.UnobservedTaskException += OnUnobservedTaskException;
+        try
+        {
+            var taskReference = CreateFailedPacketBusItem(exception);
+
+            GC.Collect();
+            GC.WaitForPendingFinalizers();
+            GC.Collect();
+
+            Assert.IsFalse(taskReference.IsAlive, "The faulted task must be collected to exercise unobserved exception reporting.");
+            Assert.AreEqual(0, unobservedExceptions);
+        }
+        finally
+        {
+            TaskScheduler.UnobservedTaskException -= OnUnobservedTaskException;
+        }
+    }
+
+    [TestMethod]
+    [DataRow(false)]
+    [DataRow(true)]
+    public async Task Wait_Failed_Packet_Bus_Item_Preserves_Exception(bool waitBeforeFailure)
+    {
+        var item = new MqttPacketBusItem(new MqttPublishPacket());
+        var exception = new MqttPendingMessagesOverflowException("test", MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage);
+        var task = waitBeforeFailure ? item.WaitAsync() : null;
+
+        item.Fail(exception);
+
+        task ??= item.WaitAsync();
+        var thrownException = await Assert.ThrowsExactlyAsync<MqttPendingMessagesOverflowException>(async () => await task);
+
+        Assert.AreSame(exception, thrownException);
+        Assert.IsTrue(task.IsFaulted);
+    }
+
     [TestMethod]
     public void Fire_Completed_Event()
     {
@@ -40,4 +94,13 @@ public sealed class MqttPacketBusItem_Tests
             await item.WaitAsync();
         });
     }
+
+    [MethodImpl(MethodImplOptions.NoInlining)]
+    static WeakReference CreateFailedPacketBusItem(Exception exception)
+    {
+        // Keep strong references out of the collecting method, including in debug builds.
+        var item = new MqttPacketBusItem(new MqttPublishPacket());
+        item.Fail(exception);
+        return new WeakReference(item.WaitAsync());
+    }
 }

Source/MQTTnet.Tests/MqttTcpChannel_Tests.cs

@@ -3,7 +3,10 @@
 // See the LICENSE file in the project root for more information.
 
 using System.Net;
+using System.Net.Security;
 using System.Net.Sockets;
+using System.Reflection;
+using System.Security.Cryptography.X509Certificates;
 using MQTTnet.Implementations;
 
 namespace MQTTnet.Tests;
@@ -12,6 +15,21 @@ namespace MQTTnet.Tests;
 [TestClass]
 public class MqttTcpChannel_Tests
 {
+    [TestMethod]
+    public void Certificate_Validation_Uses_Default_Handler_When_No_Custom_Handler()
+    {
+        var options = new MqttClientOptionsBuilder().WithTcpServer("localhost")
+            .WithTlsOptions(o => o.WithAllowUntrustedCertificates().WithIgnoreCertificateRevocationErrors())
+            .Build();
+
+        var tcpChannel = new MqttTcpChannel(options);
+
+        using var chain = new X509Chain();
+        var isValid = InvokeCertificateValidationCallback(tcpChannel, chain, SslPolicyErrors.RemoteCertificateChainErrors);
+
+        Assert.IsTrue(isValid);
+    }
+
     [TestMethod]
     public async Task Dispose_Channel_While_Used()
     {
@@ -77,4 +95,12 @@ public class MqttTcpChannel_Tests
             ct.Cancel(false);
         }
     }
-}
+
+    static bool InvokeCertificateValidationCallback(MqttTcpChannel tcpChannel, X509Chain chain, SslPolicyErrors sslPolicyErrors)
+    {
+        var method = typeof(MqttTcpChannel).GetMethod("InternalUserCertificateValidationCallback", BindingFlags.Instance | BindingFlags.NonPublic);
+        Assert.IsNotNull(method);
+
+        return (bool)method.Invoke(tcpChannel, [null, null, chain, sslPolicyErrors]);
+    }
+}

Source/MQTTnet.Tests/Server/Injection_Tests.cs

@@ -48,7 +48,7 @@ public sealed class Injection_Tests : BaseTestClass
     [TestMethod]
     public async Task Enqueue_Application_Message_At_Session_Level_QueueOverflow_DropNewMessageStrategy()
     {
-        using var testEnvironment = CreateTestEnvironment(trackUnobservedTaskException: false);
+        using var testEnvironment = CreateTestEnvironment();
 
         var server = await testEnvironment.StartServer(
             builder => builder
@@ -126,7 +126,7 @@ public sealed class Injection_Tests : BaseTestClass
     [TestMethod]
     public async Task Enqueue_Application_Message_At_Session_Level_QueueOverflow_DropOldestQueuedMessageStrategy()
     {
-        using var testEnvironment = CreateTestEnvironment(trackUnobservedTaskException: false);
+        using var testEnvironment = CreateTestEnvironment();
 
         var server = await testEnvironment.StartServer(
             builder => builder
@@ -322,7 +322,7 @@ public sealed class Injection_Tests : BaseTestClass
     [TestMethod]
     public async Task Deliver_Application_Message_At_Session_Level_QueueOverflow_DropOldestQueuedMessageStrategy()
     {
-        using var testEnvironment = CreateTestEnvironment(trackUnobservedTaskException: false);
+        using var testEnvironment = CreateTestEnvironment();
 
         var server = await testEnvironment.StartServer(
             builder => builder

Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs

@@ -476,8 +476,28 @@ public sealed class MqttV5PacketDecoder
         }
 
         var propertiesReader = new MqttV5PropertiesReader(_bufferReader);
+        var seenSingletonProperties = 0;
         while (propertiesReader.MoveNext())
         {
+            // MQTT 5.0 section 3.3.2.3 permits only User Properties and
+            // Subscription Identifiers to occur more than once in PUBLISH.
+            var propertyFlag = propertiesReader.CurrentPropertyId switch
+            {
+                MqttPropertyId.PayloadFormatIndicator => 1,
+                MqttPropertyId.MessageExpiryInterval => 2,
+                MqttPropertyId.ContentType => 4,
+                MqttPropertyId.ResponseTopic => 8,
+                MqttPropertyId.CorrelationData => 16,
+                MqttPropertyId.TopicAlias => 32,
+                _ => 0
+            };
+
+            if ((seenSingletonProperties & propertyFlag) != 0)
+            {
+                throw new MqttProtocolViolationException($"Property '{propertiesReader.CurrentPropertyId}' must not occur more than once in a PUBLISH packet.");
+            }
+
+            seenSingletonProperties |= propertyFlag;
             if (propertiesReader.CurrentPropertyId == MqttPropertyId.PayloadFormatIndicator)
             {
                 packet.PayloadFormatIndicator = propertiesReader.ReadPayloadFormatIndicator();
@@ -653,7 +673,17 @@ public sealed class MqttV5PacketDecoder
         {
             if (propertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier)
             {
+                // MQTT 5.0 section 3.8.2.1.2 forbids zero and repeated identifiers.
+                if (packet.SubscriptionIdentifier != 0)
+                {
+                    throw new MqttProtocolViolationException("Subscription Identifier must not occur more than once in a SUBSCRIBE packet.");
+                }
+
                 packet.SubscriptionIdentifier = propertiesReader.ReadSubscriptionIdentifier();
+                if (packet.SubscriptionIdentifier == 0)
+                {
+                    throw new MqttProtocolViolationException("Subscription Identifier must not be zero.");
+                }
             }
             else
             {

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -362,23 +362,18 @@ public sealed class MqttTcpChannel : IMqttChannel
 
     bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
     {
+        var eventArgs = new MqttClientCertificateValidationEventArgs(x509Certificate, chain, sslPolicyErrors, _tcpOptions);
         var certificateValidationHandler = _tcpOptions?.TlsOptions?.CertificateValidationHandler;
         if (certificateValidationHandler != null)
         {
-            var eventArgs = new MqttClientCertificateValidationEventArgs(x509Certificate, chain, sslPolicyErrors, _tcpOptions);
             return certificateValidationHandler(eventArgs);
         }
 
-        if (_tcpOptions?.TlsOptions?.IgnoreCertificateChainErrors ?? false)
+        return MqttClientDefaultCertificateValidationHandler.Handle(eventArgs);
-        {
-            sslPolicyErrors &= ~SslPolicyErrors.RemoteCertificateChainErrors;
-        }
-
-        return sslPolicyErrors == SslPolicyErrors.None;
     }
 
     X509CertificateCollection LoadCertificates()
     {
         return _tcpOptions.TlsOptions.ClientCertificatesProvider?.GetCertificates();
     }
-}
+}

Source/MQTTnet/Internal/MqttPacketBusItem.cs

@@ -33,6 +33,10 @@ public sealed class MqttPacketBusItem
     public void Fail(Exception exception)
     {
         _promise.TrySetException(exception);
+
+        // Queued packets do not always have a waiter (for example, normal server dispatch).
+        // Observe the failure here while preserving it for callers of WaitAsync.
+        _ = _promise.Task.Exception;
     }
 
     public Task<MqttPacket> WaitAsync()