improve exceptions in case of corrupt packages close pipes after exception occoured to avoid lingering connections should fix #788

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -145,6 +145,13 @@ namespace MQTTnet.AspNetCore
                     }
                 }
             }
+            catch (Exception e)
+            {
+                // completing the cannels makes sure that there is no more data read after a protocol error
+                _input?.Complete(e);
+                _output?.Complete(e);
+                throw;
+            }
             finally
             {
                 ReadingPacketCompletedCallback?.Invoke();

Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs

@@ -117,6 +117,10 @@ namespace MQTTnet.AspNetCore
             var span = _buffer.Span;
             var length = BinaryPrimitives.ReadUInt16BigEndian(span.Slice(_offset));
 
+            if (Length < _offset + length)
+            {
+                throw new MqttProtocolViolationException($"Expected at least {_offset + 2 + length} bytes but there are only {Length} bytes");
+            }
             var result = span.Slice(_offset + 2, length);
             _offset += 2 + length;
             return result;

Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs

@@ -10,6 +10,11 @@ namespace MQTTnet.AspNetCore
     {
         readonly ArrayPool<byte> _pool = ArrayPool<byte>.Create();
 
+        public SpanBasedMqttPacketWriter()
+        {
+            Reset(0);
+        }
+
         byte[] _buffer;
         int _position;
 

Source/MQTTnet/Formatter/MqttPacketBodyReader.cs

@@ -134,7 +134,7 @@ namespace MQTTnet.Formatter
         {
             if (_length < _offset + length)
             {
-                throw new ArgumentOutOfRangeException(nameof(_buffer), $"Expected at least {_offset + length} bytes but there are only {_length} bytes");
+                throw new MqttProtocolViolationException($"Expected at least {_offset + length} bytes but there are only {_length} bytes");
             }
         }
 

Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs

@@ -120,6 +120,14 @@ namespace MQTTnet.Formatter
         {
             if (receivedMqttPacket == null) throw new ArgumentNullException(nameof(receivedMqttPacket));
 
+            if (receivedMqttPacket.Body.Length < 7)
+            {
+                // 2 byte protocol name length
+                // at least 4 byte protocol name
+                // 1 byte protocol level
+                throw new MqttProtocolViolationException("Mqtt Connect packet must have at least 7 bytes");
+            }
+
             var protocolName = receivedMqttPacket.Body.ReadStringWithLengthPrefix();
             var protocolLevel = receivedMqttPacket.Body.ReadByte();
 

Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs

@@ -16,6 +16,7 @@ using MQTTnet.Formatter;
 using MQTTnet.Packets;
 using System.Net;
 using MQTTnet.AspNetCore.Extensions;
+using MQTTnet.Protocol;
 
 namespace MQTTnet.AspNetCore.Tests
 {
@@ -35,7 +36,25 @@ namespace MQTTnet.AspNetCore.Tests
 
             await Assert.ThrowsExceptionAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(TimeSpan.Zero, CancellationToken.None));
         }
-        
+
+        [TestMethod]
+        public async Task TestCorruptedConnectPacket()
+        {
+            var writer = new MqttPacketWriter();
+            var serializer = new MqttPacketFormatterAdapter(writer);
+            var pipe = new DuplexPipeMockup();
+            var connection = new DefaultConnectionContext();
+            connection.Transport = pipe;
+            var ctx = new MqttConnectionContext(serializer, connection);
+            
+            await pipe.Receive.Writer.WriteAsync(writer.AddMqttHeader(MqttControlPacketType.Connect, new byte[0]));
+
+            await Assert.ThrowsExceptionAsync<MqttProtocolViolationException>(() => ctx.ReceivePacketAsync(TimeSpan.Zero, CancellationToken.None));
+
+            // the first exception should complete the pipes so if someone tries to use the connection after that it should throw immidiatly
+            await Assert.ThrowsExceptionAsync<InvalidOperationException>(() =>  ctx.ReceivePacketAsync(TimeSpan.Zero, CancellationToken.None));
+        }
+
         [TestMethod]
         public async Task TestParallelWrites()
         {

Tests/MQTTnet.AspNetCore.Tests/SpanBasedMqttPacketWriterTests.cs

@@ -0,0 +1,15 @@
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Formatter;
+using MQTTnet.Tests;
+
+namespace MQTTnet.AspNetCore.Tests
+{
+    [TestClass]
+    public class SpanBasedMqttPacketWriterTests : MqttPacketWriter_Tests
+    {
+        protected override IMqttPacketWriter WriterFactory()
+        {
+            return new SpanBasedMqttPacketWriter();
+        }
+    }
+}

Tests/MQTTnet.Core.Tests/Extensions/MqttPacketWriterExtensions.cs

@@ -0,0 +1,15 @@
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Formatter
+{
+    public static class MqttPacketWriterExtensions
+    {
+        public static byte[] AddMqttHeader(this IMqttPacketWriter writer, MqttControlPacketType header, byte[] body)
+        {
+            writer.Write(MqttPacketWriter.BuildFixedHeader(header));
+            writer.WriteVariableLengthInteger((uint)body.Length);
+            writer.Write(body, 0, body.Length);
+            return writer.GetBuffer();
+        }
+    }
+}

Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs

@@ -6,6 +6,7 @@ using System.Text;
 using System.Threading;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Adapter;
+using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
 using MQTTnet.Formatter.V3;
 using MQTTnet.Formatter.V5;
@@ -18,6 +19,31 @@ namespace MQTTnet.Tests
     [TestClass]
     public class MqttPacketSerializer_Tests
     {
+        [TestMethod]
+        public void DetectVersionFromMqttConnectPacket()
+        {
+            var p = new MqttConnectPacket
+            {
+                ClientId = "XYZ",
+                Password = Encoding.UTF8.GetBytes("PASS"),
+                Username = "USER",
+                KeepAlivePeriod = 123,
+                CleanSession = true
+            };
+            var adapter = new MqttPacketFormatterAdapter(WriterFactory());
+                        
+            Assert.AreEqual(MqttProtocolVersion.V310, DeserializeAndDetectVersion(adapter, Serialize(p, MqttProtocolVersion.V310)));
+            Assert.AreEqual(MqttProtocolVersion.V311, DeserializeAndDetectVersion(adapter, Serialize(p, MqttProtocolVersion.V311)));
+            Assert.AreEqual(MqttProtocolVersion.V500, DeserializeAndDetectVersion(adapter, Serialize(p, MqttProtocolVersion.V500)));
+            
+            var ex = Assert.ThrowsException<MqttProtocolViolationException>(() => DeserializeAndDetectVersion(adapter, WriterFactory().AddMqttHeader(MqttControlPacketType.Connect, new byte[0])));
+            Assert.AreEqual("Mqtt Connect packet must have at least 7 bytes", ex.Message);
+            ex = Assert.ThrowsException<MqttProtocolViolationException>(() => DeserializeAndDetectVersion(adapter, WriterFactory().AddMqttHeader(MqttControlPacketType.Connect, new byte[7])));
+            Assert.AreEqual("Protocol '' not supported.", ex.Message);
+            ex = Assert.ThrowsException<MqttProtocolViolationException>(() => DeserializeAndDetectVersion(adapter, WriterFactory().AddMqttHeader(MqttControlPacketType.Connect, new byte[] { 255, 255, 0,0,0,0,0 })));
+            Assert.AreEqual("Expected at least 65537 bytes but there are only 7 bytes", ex.Message);
+        }
+
         [TestMethod]
         public void SerializeV310_MqttConnectPacket()
         {
@@ -156,7 +182,7 @@ namespace MQTTnet.Tests
         [TestMethod]
         public void Serialize_LargePacket()
         {
-            var serializer = new MqttV311PacketFormatter(new MqttPacketWriter());
+            var serializer = new MqttV311PacketFormatter(WriterFactory());
 
             const int payloadLength = 80000;
 
@@ -555,25 +581,15 @@ namespace MQTTnet.Tests
             DeserializeAndCompare(p, "sAIAew==");
         }
 
-        private static void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
+        private void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
         {
-            IMqttPacketFormatter serializer;
-            if (protocolVersion == MqttProtocolVersion.V311)
-            {
-                serializer = new MqttV311PacketFormatter(new MqttPacketWriter());
-            }
-            else if (protocolVersion == MqttProtocolVersion.V310)
-            {
-                serializer = new MqttV310PacketFormatter(new MqttPacketWriter());
-            }
-            else
-            {
-                throw new NotSupportedException();
-            }
+            Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Serialize(packet, protocolVersion)));
+        }
 
-            var data = serializer.Encode(packet);
-
-            Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(data)));
+        private byte[] Serialize(MqttBasePacket packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
+        {
+            var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, WriterFactory());
+            return Join(serializer.Encode(packet));
         }
 
         protected virtual IMqttPacketWriter WriterFactory()
@@ -635,6 +651,25 @@ namespace MQTTnet.Tests
             }
         }
 
+        private MqttProtocolVersion DeserializeAndDetectVersion(MqttPacketFormatterAdapter adapter, byte[] buffer)
+        {
+            using (var headerStream = new MemoryStream(buffer))
+            {
+                var channel = new TestMqttChannel(headerStream);
+                var fixedHeader = new byte[2];
+
+                var header = new MqttPacketReader(channel).ReadFixedHeaderAsync(fixedHeader, CancellationToken.None).GetAwaiter().GetResult().FixedHeader;
+
+                using (var bodyStream = new MemoryStream(buffer, (int)headerStream.Position, (int)header.RemainingLength))
+                {
+                    var reader = ReaderFactory(bodyStream.ToArray());
+                    var packet = new ReceivedMqttPacket(header.Flags, reader, 0);
+                    adapter.DetectProtocolVersion(packet);
+                    return adapter.ProtocolVersion;
+                }
+            }
+        }
+
         private static byte[] Join(params ArraySegment<byte>[] chunks)
         {
             var buffer = new MemoryStream();

Tests/MQTTnet.Core.Tests/MqttPacketWriter_Tests.cs

@@ -6,10 +6,15 @@ namespace MQTTnet.Tests
     [TestClass]
     public class MqttPacketWriter_Tests
     {
+        protected virtual IMqttPacketWriter WriterFactory()
+        {
+            return new MqttPacketWriter();
+        }
+
         [TestMethod]
         public void WritePacket()
         {
-            var writer = new MqttPacketWriter();
+            var writer = WriterFactory();
             Assert.AreEqual(0, writer.Length);
 
             writer.WriteWithLengthPrefix("1234567890");