From 33e4ec19438463a1b3b7cbabc916a1fdc8e4a271 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Thu, 9 Jul 2020 21:19:00 +0200 Subject: [PATCH] improve exceptions in case of corrupt packages close pipes after exception occoured to avoid lingering connections should fix #788 --- .../MqttConnectionContext.cs | 7 ++ .../SpanBasedMqttPacketBodyReader.cs | 4 ++ .../SpanBasedMqttPacketWriter.cs | 5 ++ .../MQTTnet/Formatter/MqttPacketBodyReader.cs | 2 +- .../Formatter/MqttPacketFormatterAdapter.cs | 8 +++ .../MqttConnectionContextTest.cs | 21 +++++- .../SpanBasedMqttPacketWriterTests.cs | 15 ++++ .../Extensions/MqttPacketWriterExtensions.cs | 15 ++++ .../MqttPacketSerializer_Tests.cs | 71 ++++++++++++++----- .../MqttPacketWriter_Tests.cs | 7 +- 10 files changed, 134 insertions(+), 21 deletions(-) create mode 100644 Tests/MQTTnet.AspNetCore.Tests/SpanBasedMqttPacketWriterTests.cs create mode 100644 Tests/MQTTnet.Core.Tests/Extensions/MqttPacketWriterExtensions.cs diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index 81c2cb9f..fcf329d1 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/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(); diff --git a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs index 027346fc..2f921db8 100644 --- a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs +++ b/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; diff --git a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs index 6043a618..62f678b2 100644 --- a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs +++ b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs @@ -10,6 +10,11 @@ namespace MQTTnet.AspNetCore { readonly ArrayPool _pool = ArrayPool.Create(); + public SpanBasedMqttPacketWriter() + { + Reset(0); + } + byte[] _buffer; int _position; diff --git a/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs b/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs index 1deb1fe2..ea0fa1a5 100644 --- a/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs +++ b/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"); } } diff --git a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs index 387937bd..a8dfd70d 100644 --- a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs +++ b/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(); diff --git a/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs b/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs index 46e8c651..0e545b50 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs +++ b/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(() => 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(() => 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(() => ctx.ReceivePacketAsync(TimeSpan.Zero, CancellationToken.None)); + } + [TestMethod] public async Task TestParallelWrites() { diff --git a/Tests/MQTTnet.AspNetCore.Tests/SpanBasedMqttPacketWriterTests.cs b/Tests/MQTTnet.AspNetCore.Tests/SpanBasedMqttPacketWriterTests.cs new file mode 100644 index 00000000..8e54887f --- /dev/null +++ b/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(); + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/Extensions/MqttPacketWriterExtensions.cs b/Tests/MQTTnet.Core.Tests/Extensions/MqttPacketWriterExtensions.cs new file mode 100644 index 00000000..c10ccb17 --- /dev/null +++ b/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(); + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs index 7f6cdb91..14abbdf9 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs +++ b/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(() => 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(() => DeserializeAndDetectVersion(adapter, WriterFactory().AddMqttHeader(MqttControlPacketType.Connect, new byte[7]))); + Assert.AreEqual("Protocol '' not supported.", ex.Message); + ex = Assert.ThrowsException(() => 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[] chunks) { var buffer = new MemoryStream(); diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketWriter_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketWriter_Tests.cs index bd97c1ce..ad08d52e 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketWriter_Tests.cs +++ b/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");