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
This commit is contained in:
Joseph Doherty
2026-07-24 16:30:45 -04:00
parent d421487bcd
commit 4ec01bdfc1
4 changed files with 302 additions and 22 deletions
@@ -41,9 +41,41 @@ internal sealed class MqttBrowseSession : IBrowseSession
/// <summary>Maximum characters of a payload retained as the picker's confirm-it's-the-right-topic snippet.</summary> /// <summary>Maximum characters of a payload retained as the picker's confirm-it's-the-right-topic snippet.</summary>
internal const int PayloadSnippetMaxChars = 64; internal const int PayloadSnippetMaxChars = 64;
/// <summary>
/// Hard cap on the payload bytes fed to the UTF-8 decode. A <c>#</c> 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 <i>already-known</i> 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 <see cref="PayloadSnippetMaxChars"/> × 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.
/// </summary>
internal const int PayloadInspectMaxBytes = 512;
/// <summary>
/// Cap on a whole topic name. MQTT permits ~65 KB, and the node cap bounds the node
/// <i>count</i> 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 <i>different</i>
/// topic, and the picker would commit it as a tag address.
/// </summary>
internal const int MaxTopicChars = 1024;
/// <summary>Cap on one topic segment (one tree node's label + display name).</summary>
internal const int MaxSegmentChars = 256;
/// <summary>Sentinel node id of the "results truncated" marker, mirroring <c>CapturedTreeBrowseSession</c>.</summary> /// <summary>Sentinel node id of the "results truncated" marker, mirroring <c>CapturedTreeBrowseSession</c>.</summary>
internal const string TruncatedNodeId = "__truncated__"; internal const string TruncatedNodeId = "__truncated__";
/// <summary>Sentinel node id of the "oversized topics rejected" marker.</summary>
internal const string OversizedNodeId = "__oversized__";
/// <summary>
/// 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.
/// </summary>
private static readonly UTF8Encoding Utf8Strict = new(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true);
/// <summary>Bound on the courtesy DISCONNECT issued at dispose — teardown never hangs on a dead broker.</summary> /// <summary>Bound on the courtesy DISCONNECT issued at dispose — teardown never hangs on a dead broker.</summary>
private static readonly TimeSpan DisconnectTimeout = TimeSpan.FromSeconds(5); private static readonly TimeSpan DisconnectTimeout = TimeSpan.FromSeconds(5);
@@ -60,6 +92,7 @@ internal sealed class MqttBrowseSession : IBrowseSession
private int _disposed; private int _disposed;
private long _lastUsedTicksUtc = DateTime.UtcNow.Ticks; private long _lastUsedTicksUtc = DateTime.UtcNow.Ticks;
private int _nodeCount; private int _nodeCount;
private int _oversized;
private int _publishCount; private int _publishCount;
private int _truncated; private int _truncated;
@@ -95,6 +128,14 @@ internal sealed class MqttBrowseSession : IBrowseSession
/// <summary>True once the node cap stopped recording; surfaced as a marker node from <see cref="RootAsync"/>.</summary> /// <summary>True once the node cap stopped recording; surfaced as a marker node from <see cref="RootAsync"/>.</summary>
internal bool Truncated => Volatile.Read(ref _truncated) != 0; internal bool Truncated => Volatile.Read(ref _truncated) != 0;
/// <summary>
/// True once a topic was rejected for breaching <see cref="MaxTopicChars"/> /
/// <see cref="MaxSegmentChars"/>; surfaced as its own marker node from
/// <see cref="RootAsync"/>. Kept distinct from <see cref="Truncated"/> because the operator's
/// remedy differs — narrow the prefix, versus this broker publishes absurd topic names.
/// </summary>
internal bool OversizedRejected => Volatile.Read(ref _oversized) != 0;
private bool Disposed => Volatile.Read(ref _disposed) != 0; private bool Disposed => Volatile.Read(ref _disposed) != 0;
/// <summary> /// <summary>
@@ -123,6 +164,15 @@ internal sealed class MqttBrowseSession : IBrowseSession
HasChildrenHint: false)); 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(); Touch();
return Task.FromResult<IReadOnlyList<BrowseNode>>(nodes); return Task.FromResult<IReadOnlyList<BrowseNode>>(nodes);
} }
@@ -209,7 +259,24 @@ internal sealed class MqttBrowseSession : IBrowseSession
{ {
if (Disposed || string.IsNullOrWhiteSpace(topic)) return; if (Disposed || string.IsNullOrWhiteSpace(topic)) return;
if (topic.Length > MaxTopicChars)
{
Volatile.Write(ref _oversized, 1);
return;
}
var segments = topic.Split('/'); 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; var level = _roots;
TopicNode? node = null; TopicNode? node = null;
var path = string.Empty; var path = string.Empty;
@@ -270,6 +337,10 @@ internal sealed class MqttBrowseSession : IBrowseSession
/// <summary> /// <summary>
/// Best-effort type inference from a payload body. Explicit authoring stays the authority /// 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. /// (design §4) — this only decorates the picker. Non-UTF-8 bodies are reported as opaque.
/// <para>
/// Runs on MQTTnet's dispatcher thread, so the work is bounded before any decode: at most
/// <see cref="PayloadInspectMaxBytes"/> bytes are ever examined, however large the message.
/// </para>
/// </summary> /// </summary>
/// <param name="payload">The raw message body.</param> /// <param name="payload">The raw message body.</param>
/// <returns>The inferred type and a bounded display snippet.</returns> /// <returns>The inferred type and a bounded display snippet.</returns>
@@ -277,37 +348,102 @@ internal sealed class MqttBrowseSession : IBrowseSession
{ {
if (payload.IsEmpty) return new ObservedValue(DriverDataType.String, "(empty)"); 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; string text;
try try
{ {
// Throw-on-invalid so genuinely binary payloads are reported as such rather than text = Utf8Strict.GetString(window);
// silently rendered as a field of replacement characters.
text = new UTF8Encoding(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true)
.GetString(payload);
} }
catch (DecoderFallbackException) catch (DecoderFallbackException)
{ {
return new ObservedValue(DriverDataType.String, $"({payload.Length} bytes, binary)"); return new ObservedValue(DriverDataType.String, $"({totalBytes} bytes, binary)");
} }
var trimmed = text.Trim(); 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 // 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. // 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));
}
/// <summary>
/// Renders decoded payload text as a single-line, length-bounded snippet. Control characters
/// (newlines, tabs, ANSI escapes) become spaces — <see cref="string.Trim()"/> only strips them
/// at the ends, and an embedded one would break the picker's single-line attribute row.
/// </summary>
/// <param name="text">The trimmed decoded text.</param>
/// <param name="clipped">Whether bytes beyond the inspection window were discarded.</param>
/// <returns>The display snippet.</returns>
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);
}
/// <summary>Replaces control characters with spaces and trims the trailing edge.</summary>
/// <param name="span">The characters to scrub.</param>
/// <returns>The scrubbed text.</returns>
private static string Scrub(ReadOnlySpan<char> span)
{
Span<char> 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();
}
/// <summary>
/// Shortens a byte window so it ends on a whole UTF-8 sequence. Walks back over continuation
/// bytes (<c>10xxxxxx</c>) 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.
/// </summary>
/// <param name="bytes">The blindly-sliced window.</param>
/// <returns>The window, shortened to a sequence boundary.</returns>
private static ReadOnlySpan<byte> TrimToUtf8Boundary(ReadOnlySpan<byte> 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, i--;
_ when long.TryParse(trimmed, NumberStyles.Integer, CultureInfo.InvariantCulture, out _) walked++;
=> DriverDataType.Int64, }
_ when double.TryParse(trimmed, NumberStyles.Float, CultureInfo.InvariantCulture, out _)
=> DriverDataType.Float64, if (i < 0) return ReadOnlySpan<byte>.Empty; // nothing but continuation bytes
_ => DriverDataType.String,
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];
} }
/// <summary>Projects one level of the tree into browse nodes, ordered so the picker is stable.</summary> /// <summary>Projects one level of the tree into browse nodes, ordered so the picker is stable.</summary>
@@ -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 /// publishes: an operator opening a picker must not be able to inject a message into a
/// running plant. /// running plant.
/// </para> /// </para>
/// <para>
/// <b>Layering note.</b> Unlike every other bespoke browser here, this project references the
/// runtime <c>.Driver</c> project (to reuse <c>MqttConnection.BuildClientOptions</c> 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 <c>ProjectReference</c> in
/// <c>ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj</c> before copying this project as a
/// template.
/// </para>
/// </summary> /// </summary>
public sealed class MqttDriverBrowser : IDriverBrowser public sealed class MqttDriverBrowser : IDriverBrowser
{ {
@@ -14,12 +14,32 @@
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Commons\ZB.MOM.WW.OtOpcUa.Commons.csproj"/> <ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Commons\ZB.MOM.WW.OtOpcUa.Commons.csproj"/>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj"/> <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj"/>
<!-- <!--
Deviation from the plan's ref list (which named .Contracts only): the browse connect ┌──────────────────────────────────────────────────────────────────────────────────┐
reuses MqttConnection.BuildClientOptions rather than reimplementing it. That method │ KNOWN, DELIBERATE EXCEPTION to the .Browser → .Contracts pattern. │
owns the TLS posture, the CA-pin chain validator and the credential handling; a second │ DO NOT COPY THIS LINE INTO A NEW DRIVER'S BROWSER WITHOUT READING THIS. │
copy here would be a security divergence waiting to happen (browse silently not └──────────────────────────────────────────────────────────────────────────────────┘
honouring a pinned CA while the runtime driver does). It is pure — no I/O, no network —
so the dependency costs nothing at browse time. Every other bespoke browser in this repo (OpcUaClient, Galaxy) references only its own
.Contracts project plus its transport package. This one also references the runtime
.Driver project. That is a real widening, not a free one, and it was accepted knowingly:
WHY: the browse CONNECT reuses MqttConnection.BuildClientOptions, which owns the TLS
posture, the PEM CA-pin chain validator, the credential handling and the protocol
version mapping (~100 lines). A second copy here would be a silent security
divergence — browse quietly not honouring a pinned CA while the runtime driver does.
The method is pure (no I/O, no network), so the dependency costs nothing at run time.
WHAT IT COSTS: the AdminUI's project graph does NOT otherwise include Core.csproj
(only Host.csproj and the runtime Driver.* projects do). Referencing this browser
from the AdminUI therefore pulls in Core + Polly.Core + Serilog at build/deploy time.
THE CLEAN FIX (deferred, not blocked): a small leaf project — say
ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport (net10, refs .Contracts + MQTTnet) — holding
BuildClientOptions/ConfigureTls/the CA-pin helpers, referenced by BOTH .Driver and
.Browser. Note the obvious-looking alternative does NOT work: those members cannot
move into .Contracts, because .Contracts is deliberately transport-free (Task 1
removed its MQTTnet reference to keep it so) and BuildClientOptions returns
MqttClientOptions, an MQTTnet type.
--> -->
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj"/> <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj"/>
</ItemGroup> </ItemGroup>
@@ -248,6 +248,122 @@ public sealed class MqttBrowseSessionTests
root.ShouldContain(n => n.NodeId == MqttBrowseSession.TruncatedNodeId); 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 // ---------------------------------------------------------------- the browser
[Fact] [Fact]