From 4ec01bdfc15997d888eae08b2a8b7bc561ae58fe Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 24 Jul 2026 16:30:45 -0400 Subject: [PATCH] fix(mqtt): bound browse observation against a chatty or malicious broker Three review findings, all "unbounded work/memory from untrusted broker input" rather than correctness bugs - but this points at a live plant broker. 1. InferPayload decoded the WHOLE payload on MQTTnet's dispatcher thread before trimming to 64 chars. The node cap never engages here: a multi-MB blob on an already-known topic creates no node, it just updates one, so the cost repeats per message forever. Now at most PayloadInspectMaxBytes (512) are examined. A blind slice can cut a multi-byte sequence, which the strict decoder would report as "binary" - TrimToUtf8Boundary backs the window off to the last whole sequence so good UTF-8 is never mislabelled. A clipped payload is String by construction rather than inferred from a partial view. 2. The node cap bounded COUNT, not bytes. MQTT topics run to ~65KB, so 50k nodes near that ceiling is a multi-GB tree. Topics now bounded at 1024 chars and segments at 256, rejected whole (a truncated topic is a DIFFERENT topic the picker would commit as a tag address) and surfaced via an __oversized__ marker kept distinct from __truncated__ - the operator's remedy differs. 3. Control characters in a snippet would break the picker's single-line row; Trim() only strips the ends. Now scrubbed to spaces. Each guard was falsified independently by neutering it and confirming exactly one test reddens. That found a real gap: the oversized-topic test was passing via the SEGMENT bound, leaving the whole-topic bound untested - the test now uses many short segments so only the topic bound can reject it. Also records the .Driver project reference as a known, deliberate exception to the .Browser -> .Contracts pattern, with its cost (AdminUI gains Core + Polly + Serilog) and the clean fix (a leaf transport project; NOT a move into .Contracts, which is deliberately transport-free). Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- .../MqttBrowseSession.cs | 168 ++++++++++++++++-- .../MqttDriverBrowser.cs | 8 + ....MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj | 32 +++- .../MqttBrowseSessionTests.cs | 116 ++++++++++++ 4 files changed, 302 insertions(+), 22 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs index eca93d61..3a393310 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs @@ -41,9 +41,41 @@ internal sealed class MqttBrowseSession : IBrowseSession /// Maximum characters of a payload retained as the picker's confirm-it's-the-right-topic snippet. internal const int PayloadSnippetMaxChars = 64; + /// + /// Hard cap on the payload bytes fed to the UTF-8 decode. A # subscription receives + /// whatever the broker has — multi-MB retained blobs and JSON dumps included — and the node + /// cap does not help: a large payload on an already-known topic creates no node, it + /// just updates one, so an uncapped decode is unbounded work on MQTTnet's pump thread on + /// every single message, forever. 512 is × 4 (UTF-8's + /// worst-case bytes per char, so the snippet can always be filled) doubled, leaving headroom + /// for leading whitespace ahead of a scalar. + /// + internal const int PayloadInspectMaxBytes = 512; + + /// + /// Cap on a whole topic name. MQTT permits ~65 KB, and the node cap bounds the node + /// count only — 50 000 nodes near that ceiling is a multi-GB tree. Topics past this + /// are rejected outright rather than truncated: a truncated topic is a different + /// topic, and the picker would commit it as a tag address. + /// + internal const int MaxTopicChars = 1024; + + /// Cap on one topic segment (one tree node's label + display name). + internal const int MaxSegmentChars = 256; + /// Sentinel node id of the "results truncated" marker, mirroring CapturedTreeBrowseSession. internal const string TruncatedNodeId = "__truncated__"; + /// Sentinel node id of the "oversized topics rejected" marker. + internal const string OversizedNodeId = "__oversized__"; + + /// + /// Throw-on-invalid so genuinely binary payloads are reported as such rather than silently + /// rendered as a field of replacement characters. Cached: allocating an encoding per message + /// would be per-message garbage on the dispatcher thread. + /// + private static readonly UTF8Encoding Utf8Strict = new(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true); + /// Bound on the courtesy DISCONNECT issued at dispose — teardown never hangs on a dead broker. private static readonly TimeSpan DisconnectTimeout = TimeSpan.FromSeconds(5); @@ -60,6 +92,7 @@ internal sealed class MqttBrowseSession : IBrowseSession private int _disposed; private long _lastUsedTicksUtc = DateTime.UtcNow.Ticks; private int _nodeCount; + private int _oversized; private int _publishCount; private int _truncated; @@ -95,6 +128,14 @@ internal sealed class MqttBrowseSession : IBrowseSession /// True once the node cap stopped recording; surfaced as a marker node from . internal bool Truncated => Volatile.Read(ref _truncated) != 0; + /// + /// True once a topic was rejected for breaching / + /// ; surfaced as its own marker node from + /// . Kept distinct from because the operator's + /// remedy differs — narrow the prefix, versus this broker publishes absurd topic names. + /// + internal bool OversizedRejected => Volatile.Read(ref _oversized) != 0; + private bool Disposed => Volatile.Read(ref _disposed) != 0; /// @@ -123,6 +164,15 @@ internal sealed class MqttBrowseSession : IBrowseSession HasChildrenHint: false)); } + if (OversizedRejected) + { + nodes.Add(new BrowseNode( + OversizedNodeId, + "⚠ Some topics were too long to show — use manual entry for those", + BrowseNodeKind.Folder, + HasChildrenHint: false)); + } + Touch(); return Task.FromResult>(nodes); } @@ -209,7 +259,24 @@ internal sealed class MqttBrowseSession : IBrowseSession { if (Disposed || string.IsNullOrWhiteSpace(topic)) return; + if (topic.Length > MaxTopicChars) + { + Volatile.Write(ref _oversized, 1); + return; + } + var segments = topic.Split('/'); + + // Checked as a pre-pass so a rejected topic leaves no half-built path behind. + foreach (var segment in segments) + { + if (segment.Length > MaxSegmentChars) + { + Volatile.Write(ref _oversized, 1); + return; + } + } + var level = _roots; TopicNode? node = null; var path = string.Empty; @@ -270,6 +337,10 @@ internal sealed class MqttBrowseSession : IBrowseSession /// /// Best-effort type inference from a payload body. Explicit authoring stays the authority /// (design §4) — this only decorates the picker. Non-UTF-8 bodies are reported as opaque. + /// + /// Runs on MQTTnet's dispatcher thread, so the work is bounded before any decode: at most + /// bytes are ever examined, however large the message. + /// /// /// The raw message body. /// The inferred type and a bounded display snippet. @@ -277,37 +348,102 @@ internal sealed class MqttBrowseSession : IBrowseSession { if (payload.IsEmpty) return new ObservedValue(DriverDataType.String, "(empty)"); + var totalBytes = payload.Length; + var clipped = totalBytes > PayloadInspectMaxBytes; + // A blind slice can cut a multi-byte sequence in half, which the strict decoder would report + // as "binary" — mislabelling a perfectly good UTF-8 payload. Back the window off to the last + // whole sequence instead. + var window = clipped ? TrimToUtf8Boundary(payload[..PayloadInspectMaxBytes]) : payload; + string text; try { - // Throw-on-invalid so genuinely binary payloads are reported as such rather than - // silently rendered as a field of replacement characters. - text = new UTF8Encoding(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true) - .GetString(payload); + text = Utf8Strict.GetString(window); } catch (DecoderFallbackException) { - return new ObservedValue(DriverDataType.String, $"({payload.Length} bytes, binary)"); + return new ObservedValue(DriverDataType.String, $"({totalBytes} bytes, binary)"); } var trimmed = text.Trim(); - var snippet = trimmed.Length > PayloadSnippetMaxChars - ? string.Concat(trimmed.AsSpan(0, PayloadSnippetMaxChars), "…") - : trimmed; // Ordered narrowest-first. "1" infers Int64 rather than Boolean: an integer reading is the // commoner meaning and, unlike the reverse, does not silently collapse a range to two states. - var dataType = trimmed switch + // A clipped payload is never a scalar, and inferring from a partial view could read a + // whitespace-padded prefix as a number it is not — so it is String by construction. + var dataType = clipped + ? DriverDataType.String + : trimmed switch + { + _ when bool.TryParse(trimmed, out _) => DriverDataType.Boolean, + _ when long.TryParse(trimmed, NumberStyles.Integer, CultureInfo.InvariantCulture, out _) + => DriverDataType.Int64, + _ when double.TryParse(trimmed, NumberStyles.Float, CultureInfo.InvariantCulture, out _) + => DriverDataType.Float64, + _ => DriverDataType.String, + }; + + return new ObservedValue(dataType, Sanitize(trimmed, clipped)); + } + + /// + /// Renders decoded payload text as a single-line, length-bounded snippet. Control characters + /// (newlines, tabs, ANSI escapes) become spaces — only strips them + /// at the ends, and an embedded one would break the picker's single-line attribute row. + /// + /// The trimmed decoded text. + /// Whether bytes beyond the inspection window were discarded. + /// The display snippet. + private static string Sanitize(string text, bool clipped) + { + var overLength = text.Length > PayloadSnippetMaxChars; + var span = overLength ? text.AsSpan(0, PayloadSnippetMaxChars) : text.AsSpan(); + + // clipped, not just overLength: a huge payload whose first 512 bytes are mostly whitespace + // yields a short snippet that would otherwise claim to be the whole message. + return overLength || clipped ? Scrub(span) + "…" : Scrub(span); + } + + /// Replaces control characters with spaces and trims the trailing edge. + /// The characters to scrub. + /// The scrubbed text. + private static string Scrub(ReadOnlySpan span) + { + Span buffer = span.Length <= 128 ? stackalloc char[span.Length] : new char[span.Length]; + for (var i = 0; i < span.Length; i++) buffer[i] = char.IsControl(span[i]) ? ' ' : span[i]; + return new string(buffer).TrimEnd(); + } + + /// + /// Shortens a byte window so it ends on a whole UTF-8 sequence. Walks back over continuation + /// bytes (10xxxxxx) to the lead byte and drops it when the sequence it starts runs past + /// the window. A stray continuation byte is left in place — that is genuine corruption, and the + /// strict decoder should say so. + /// + /// The blindly-sliced window. + /// The window, shortened to a sequence boundary. + private static ReadOnlySpan TrimToUtf8Boundary(ReadOnlySpan bytes) + { + var i = bytes.Length - 1; + var walked = 0; + while (i >= 0 && (bytes[i] & 0xC0) == 0x80 && walked < 3) { - _ when bool.TryParse(trimmed, out _) => DriverDataType.Boolean, - _ when long.TryParse(trimmed, NumberStyles.Integer, CultureInfo.InvariantCulture, out _) - => DriverDataType.Int64, - _ when double.TryParse(trimmed, NumberStyles.Float, CultureInfo.InvariantCulture, out _) - => DriverDataType.Float64, - _ => DriverDataType.String, + i--; + walked++; + } + + if (i < 0) return ReadOnlySpan.Empty; // nothing but continuation bytes + + var needed = bytes[i] switch + { + < 0x80 => 1, // ASCII + >= 0xF0 => 4, + >= 0xE0 => 3, + >= 0xC0 => 2, + _ => 1, // stray continuation byte — let the decoder reject it }; - return new ObservedValue(dataType, snippet); + return bytes.Length - i >= needed ? bytes : bytes[..i]; } /// Projects one level of the tree into browse nodes, ordered so the picker is stable. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs index ee87b9b4..ce2f3366 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs @@ -19,6 +19,14 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser; /// publishes: an operator opening a picker must not be able to inject a message into a /// running plant. /// +/// +/// Layering note. Unlike every other bespoke browser here, this project references the +/// runtime .Driver project (to reuse MqttConnection.BuildClientOptions rather +/// than fork the TLS / CA-pin path). That is a known, deliberate exception with a documented +/// cost and a documented clean fix — see the comment on that ProjectReference in +/// ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj before copying this project as a +/// template. +/// /// public sealed class MqttDriverBrowser : IDriverBrowser { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj index bdc8e8c6..e32fa07f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj @@ -14,12 +14,32 @@ diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs index b2a309c9..bff929cc 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs @@ -248,6 +248,122 @@ public sealed class MqttBrowseSessionTests root.ShouldContain(n => n.NodeId == MqttBrowseSession.TruncatedNodeId); } + [Fact] + public async Task OversizedTopic_IsRejectedWhole_AndSurfacesAMarker() + { + // A truncated topic is a DIFFERENT topic — the picker would commit it as a tag address. + // Every segment here is 4 chars, well under MaxSegmentChars, so ONLY the whole-topic bound + // can reject it — the segment guard must not be what makes this test pass. + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("good/topic", "1"); + + var manyShortSegments = "huge/" + string.Join('/', Enumerable.Repeat("abc", MqttBrowseSession.MaxTopicChars / 4 + 8)); + manyShortSegments.Length.ShouldBeGreaterThan(MqttBrowseSession.MaxTopicChars); + manyShortSegments.Split('/').ShouldAllBe(seg => seg.Length <= MqttBrowseSession.MaxSegmentChars); + s.ObserveTopicForTest(manyShortSegments, "1"); + + var root = await s.RootAsync(TestContext.Current.CancellationToken); + root.ShouldContain(n => n.NodeId == MqttBrowseSession.OversizedNodeId); + root.ShouldContain(n => n.DisplayName == "good"); + root.ShouldNotContain(n => n.DisplayName == "huge"); // no half-built path left behind + } + + [Fact] + public async Task OversizedSegment_IsRejectedWhole_EvenWhenTheTopicFits() + { + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("a/" + new string('y', MqttBrowseSession.MaxSegmentChars + 1) + "/b", "1"); + + s.OversizedRejected.ShouldBeTrue(); + (await s.ExpandAsync("a", TestContext.Current.CancellationToken)).ShouldBeEmpty(); + var root = await s.RootAsync(TestContext.Current.CancellationToken); + root.ShouldNotContain(n => n.DisplayName == "a"); + } + + [Fact] + public async Task TopicAtExactlyTheCap_IsAccepted() + { + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest(new string('z', MqttBrowseSession.MaxSegmentChars), "1"); + + s.OversizedRejected.ShouldBeFalse(); + (await s.RootAsync(TestContext.Current.CancellationToken)).ShouldHaveSingleItem(); + } + + // ---------------------------------------------------------------- payload bounds + + [Fact] + public async Task PayloadBeyondTheInspectionWindow_IsNeverExamined() + { + // The discriminating test for the byte cap: content past PayloadInspectMaxBytes cannot + // influence the result at all. With an unbounded decode the trailing scalar would be found + // and typed Int64, and would show up in the snippet. + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest( + "t/tail", + new string(' ', MqttBrowseSession.PayloadInspectMaxBytes + 64) + "42"); + + var a = (await s.AttributesAsync("t/tail", TestContext.Current.CancellationToken)).ShouldHaveSingleItem(); + a.DriverDataType.ShouldBe(nameof(Core.Abstractions.DriverDataType.String)); + a.SecurityClass.ShouldNotContain("42"); + } + + [Fact] + public async Task HugePayload_YieldsABoundedSnippetMarkedAsClipped() + { + // A '#' subscription receives multi-MB retained blobs; the node cap never engages for a + // large payload on an already-known topic, so the decode itself must be bounded. + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("t/blob", new string('a', 5_000_000)); + + var a = (await s.AttributesAsync("t/blob", TestContext.Current.CancellationToken)).ShouldHaveSingleItem(); + a.SecurityClass.Length.ShouldBeLessThanOrEqualTo(MqttBrowseSession.PayloadSnippetMaxChars + 1); + a.SecurityClass.ShouldEndWith("…"); + a.DriverDataType.ShouldBe(nameof(Core.Abstractions.DriverDataType.String)); + } + + [Fact] + public async Task ClippedMultiByteUtf8_IsNotMislabelledAsBinary() + { + // The inspection window can land mid-sequence; backing off to a whole sequence keeps a + // perfectly good UTF-8 payload from being reported as binary. + await using var s = new MqttBrowseSession(MqttMode.Plain); + + // 3 bytes per char, so 512 / 3 leaves a partial sequence at the window edge. + s.ObserveTopicForTest("t/utf8", new string('☃', 4000)); + + var a = (await s.AttributesAsync("t/utf8", TestContext.Current.CancellationToken)).ShouldHaveSingleItem(); + a.SecurityClass.ShouldStartWith("☃"); + a.SecurityClass.ShouldNotContain("binary"); + a.SecurityClass.ShouldNotContain("�"); // no replacement char from a split sequence + } + + [Fact] + public async Task BinaryPayload_IsReportedAsOpaque_WithItsTrueLength() + { + await using var s = new MqttBrowseSession(MqttMode.Plain); + var binary = new byte[2048]; + Array.Fill(binary, (byte)0xC0); // never legal UTF-8 + s.Observe("t/bin", binary); + + var a = (await s.AttributesAsync("t/bin", TestContext.Current.CancellationToken)).ShouldHaveSingleItem(); + a.SecurityClass.ShouldBe("(2048 bytes, binary)"); // full length, not the clipped window + } + + [Fact] + public async Task EmbeddedControlCharacters_AreScrubbedToSpaces() + { + // Trim() only strips the ends; an embedded newline would break the picker's single-line row. + await using var s = new MqttBrowseSession(MqttMode.Plain); + s.ObserveTopicForTest("t/multiline", "line one\nline two\ttabbed"); + + var a = (await s.AttributesAsync("t/multiline", TestContext.Current.CancellationToken)).ShouldHaveSingleItem(); + a.SecurityClass.ShouldNotContain("\n"); + a.SecurityClass.ShouldNotContain("\t"); + a.SecurityClass.ShouldNotContain(""); + a.SecurityClass.ShouldStartWith("line one line two tabbed"); + } + // ---------------------------------------------------------------- the browser [Fact]