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[31m");
+
+ 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]