feat(mqtt): MqttDriverProbe CONNECT handshake
Test-connect probe for the Mqtt driver type: parses MqttDriverOptions
(shared enum-as-name JsonSerializerOptions), opens a bounded MQTT CONNECT
via MqttConnection.BuildClientOptions with a distinct -probe-{guid8}
client-id suffix, and classifies the outcome. Live-probed against
MQTTnet 5.2.0.1603 to confirm the real contract: a broker-rejected
CONNACK is a MqttClientConnectResult.ResultCode, never a thrown
exception, and timeout classification checks the deadline token itself
rather than switching on exception type (which varies unpredictably
between MqttConnectingFailedException and bare OperationCanceledException
for the same frozen-peer shape).
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,244 @@
|
||||
using System.Diagnostics;
|
||||
using System.Net.Sockets;
|
||||
using System.Security.Authentication;
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Serialization;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using MQTTnet;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
|
||||
|
||||
/// <summary>
|
||||
/// CONNECT-handshake probe for the <see cref="MqttDriverOptions"/>-shaped driver config.
|
||||
/// Opens a bounded MQTT CONNECT against the configured broker; a CONNACK-accepted response
|
||||
/// (<c>MqttClientConnectResultCode.Success</c>) is the "device is answering" proof (green +
|
||||
/// latency). Refused/TLS/auth/timeout failures each surface a targeted message. Mirrors
|
||||
/// <c>ModbusDriverProbe</c>'s shape.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Reuses <see cref="MqttConnection.BuildClientOptions"/></b> for the connect (TLS,
|
||||
/// CA-pin, credentials, protocol-version mapping) rather than hand-rolling a second
|
||||
/// connect path — a duplicate would be a security divergence. A distinct client-id suffix
|
||||
/// (see <see cref="ProbeClientIdPrefix"/>) keeps a probe from colliding with the driver's
|
||||
/// own session: an MQTT broker disconnects an existing client when a new one CONNECTs
|
||||
/// with the same client id, so a probe that reused the driver's id would knock the
|
||||
/// running driver offline every time an operator clicked "Test connect".
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>A broker-rejected CONNACK is not a thrown exception.</b> Live-probed against
|
||||
/// MQTTnet 5.2.0.1603: <c>IMqttClient.ConnectAsync</c> returns a
|
||||
/// <c>MqttClientConnectResult</c> whose <c>ResultCode</c> carries the broker's CONNACK
|
||||
/// reason (e.g. <c>NotAuthorized</c>, <c>BadUserNameOrPassword</c>) — it does
|
||||
/// <b>not</b> throw for a rejected CONNACK. Only transport-level failures throw, and
|
||||
/// their shape varies by where the failure occurs: a closed/refused port throws
|
||||
/// <c>MqttCommunicationException</c> wrapping a <see cref="SocketException"/>; an
|
||||
/// untrusted broker certificate throws <c>MqttCommunicationException</c> wrapping an
|
||||
/// <see cref="AuthenticationException"/>; a broker that accepts the TCP handshake and
|
||||
/// then never answers CONNACK throws either <c>MqttConnectingFailedException</c>
|
||||
/// (wrapping "Connection closed") or a bare <see cref="OperationCanceledException"/>
|
||||
/// ("MQTT connect canceled"), inconsistently, depending on whether MQTTnet's own
|
||||
/// internal <c>Timeout</c> or the deadline token wins the race. Consequently this probe —
|
||||
/// like <see cref="MqttConnection.Classify"/> — never switches on exception type to
|
||||
/// detect a timeout; it checks the deadline token itself, which is authoritative
|
||||
/// regardless of which exception shape MQTTnet happened to throw.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class MqttDriverProbe : IDriverProbe
|
||||
{
|
||||
/// <summary>
|
||||
/// Marks the transient probe identity in the broker's client-id/session logs — mirrors the
|
||||
/// browser's own <c>-browse-{guid8}</c> suffix so the two transient sessions are
|
||||
/// distinguishable in broker-side logs, and so a probe can never collide with (and knock
|
||||
/// offline) the driver's own live session.
|
||||
/// </summary>
|
||||
internal const string ProbeClientIdPrefix = "-probe-";
|
||||
|
||||
/// <summary>
|
||||
/// Shared JSON options for every MQTT config-parsing seam in this driver: enums
|
||||
/// (<see cref="MqttMode"/>, <see cref="MqttPayloadFormat"/>, <c>protocolVersion</c>)
|
||||
/// round-trip by <b>name</b>, never ordinal — the repo's documented enum-serialization
|
||||
/// trap (AdminUI-authored configs with numeric enum fields fault the driver). Task 9's
|
||||
/// <c>MqttDriverFactoryExtensions</c> is expected to converge on this same instance
|
||||
/// rather than defining its own.
|
||||
/// </summary>
|
||||
internal static readonly JsonSerializerOptions JsonOpts = new()
|
||||
{
|
||||
PropertyNameCaseInsensitive = true,
|
||||
UnmappedMemberHandling = JsonUnmappedMemberHandling.Skip,
|
||||
Converters = { new JsonStringEnumConverter() },
|
||||
};
|
||||
|
||||
/// <inheritdoc />
|
||||
// Literal rather than a constant: DriverTypeNames.Mqtt does not exist yet — Task 9 adds it, and
|
||||
// owns that file. The string must stay EXACTLY "Mqtt": a DriverType string that drifts from the
|
||||
// persisted one silently breaks driver dispatch (the repo's ModbusTcp/Modbus incident).
|
||||
public string DriverType => "Mqtt";
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<DriverProbeResult> ProbeAsync(string configJson, TimeSpan timeout, CancellationToken ct)
|
||||
{
|
||||
MqttDriverOptions? options;
|
||||
try
|
||||
{
|
||||
options = JsonSerializer.Deserialize<MqttDriverOptions>(configJson, JsonOpts);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return new DriverProbeResult(false, $"Config JSON is invalid: {ex.Message}", null);
|
||||
}
|
||||
|
||||
if (options is null)
|
||||
{
|
||||
return new DriverProbeResult(false, "Config JSON deserialized to null.", null);
|
||||
}
|
||||
|
||||
if (string.IsNullOrWhiteSpace(options.Host) || options.Port is <= 0 or > 65535)
|
||||
{
|
||||
return new DriverProbeResult(false, "Config has no host/port to probe.", null);
|
||||
}
|
||||
|
||||
// Honour the config's own connectTimeoutSeconds (factory parity, mirrors
|
||||
// ModbusDriverProbe) — the caller's timeout is the fallback when the config omits it.
|
||||
var effectiveSeconds = options.ConnectTimeoutSeconds > 0
|
||||
? options.ConnectTimeoutSeconds
|
||||
: (int)Math.Ceiling(timeout.TotalSeconds);
|
||||
if (effectiveSeconds <= 0)
|
||||
{
|
||||
effectiveSeconds = 1;
|
||||
}
|
||||
|
||||
// Rebuild with the resolved timeout so BuildClientOptions' own WithTimeout(...) and this
|
||||
// method's deadline agree — both must expire at the same instant for the deadline check
|
||||
// below to be a reliable signal regardless of which of the two actually threw.
|
||||
var probeOptions = options with { ConnectTimeoutSeconds = effectiveSeconds };
|
||||
|
||||
MqttClientOptions clientOptions;
|
||||
try
|
||||
{
|
||||
clientOptions = MqttConnection.BuildClientOptions(
|
||||
probeOptions,
|
||||
ProbeClientIdPrefix + Guid.NewGuid().ToString("N")[..8],
|
||||
NullLogger.Instance);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Never let a configured Password reach the message — BuildClientOptions itself never
|
||||
// logs it, and no exception path through it can format credential values.
|
||||
return new DriverProbeResult(false, $"Config could not be used to build a connection: {ex.Message}", null);
|
||||
}
|
||||
|
||||
var target = $"{probeOptions.Host}:{probeOptions.Port}";
|
||||
var sw = Stopwatch.StartNew();
|
||||
|
||||
using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(effectiveSeconds));
|
||||
using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, deadline.Token);
|
||||
|
||||
using var client = new MqttClientFactory().CreateMqttClient();
|
||||
|
||||
MqttClientConnectResult result;
|
||||
try
|
||||
{
|
||||
result = await client.ConnectAsync(clientOptions, linked.Token).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return new DriverProbeResult(false, Classify(target, ex, ct, deadline, effectiveSeconds), null);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
if (result.ResultCode != MqttClientConnectResultCode.Success)
|
||||
{
|
||||
return new DriverProbeResult(false, ClassifyRejectedConnack(target, result.ResultCode), null);
|
||||
}
|
||||
|
||||
sw.Stop();
|
||||
return new DriverProbeResult(true, "MQTT CONNECT OK", sw.Elapsed);
|
||||
}
|
||||
finally
|
||||
{
|
||||
await DisconnectBestEffortAsync(client, effectiveSeconds).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Maps a broker-rejected CONNACK (<see cref="MqttClientConnectResult.ResultCode"/> not
|
||||
/// <c>Success</c> — never an exception, see the type remarks) to a targeted, credential-free
|
||||
/// message.
|
||||
/// </summary>
|
||||
private static string ClassifyRejectedConnack(string target, MqttClientConnectResultCode code) => code switch
|
||||
{
|
||||
MqttClientConnectResultCode.NotAuthorized or MqttClientConnectResultCode.BadUserNameOrPassword
|
||||
=> $"Broker at {target} rejected the credentials ({code}).",
|
||||
MqttClientConnectResultCode.UnsupportedProtocolVersion
|
||||
=> $"Broker at {target} does not support the configured protocol version ({code}).",
|
||||
MqttClientConnectResultCode.Banned
|
||||
=> $"Broker at {target} has banned this client ({code}).",
|
||||
_ => $"Broker at {target} refused CONNECT: {code}.",
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// Classifies a thrown connect failure. Precedence mirrors
|
||||
/// <see cref="MqttConnection.Classify"/>: caller cancellation, then the deadline, then the
|
||||
/// exception's own shape — the deadline check is authoritative regardless of which exception
|
||||
/// type MQTTnet happened to throw (see the type remarks for why exception type alone is not
|
||||
/// reliable here).
|
||||
/// </summary>
|
||||
private static string Classify(
|
||||
string target,
|
||||
Exception ex,
|
||||
CancellationToken callerToken,
|
||||
CancellationTokenSource deadline,
|
||||
int effectiveSeconds)
|
||||
{
|
||||
if (callerToken.IsCancellationRequested)
|
||||
{
|
||||
return "Probe was cancelled.";
|
||||
}
|
||||
|
||||
if (deadline.IsCancellationRequested)
|
||||
{
|
||||
return $"Probe timed out after {effectiveSeconds}s.";
|
||||
}
|
||||
|
||||
// Walk to the root cause — MQTTnet wraps transport/TLS failures one or two levels deep.
|
||||
var root = ex;
|
||||
while (root.InnerException is not null)
|
||||
{
|
||||
root = root.InnerException;
|
||||
}
|
||||
|
||||
return root switch
|
||||
{
|
||||
SocketException se => $"Connect to {target} failed: {se.SocketErrorCode}",
|
||||
AuthenticationException => $"TLS handshake with {target} failed: {root.Message}",
|
||||
_ => $"Connect to {target} failed: {root.Message}",
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Best-effort teardown of a session this probe established. Failure here is not the
|
||||
/// probe's business — the CONNECT outcome already decided Ok/failure — and it must never be
|
||||
/// allowed to hang past the connect budget.
|
||||
/// </summary>
|
||||
private static async Task DisconnectBestEffortAsync(IMqttClient client, int timeoutSeconds)
|
||||
{
|
||||
if (!client.IsConnected)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
using var disconnectDeadline = new CancellationTokenSource(TimeSpan.FromSeconds(timeoutSeconds));
|
||||
await client.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectDeadline.Token)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
catch
|
||||
{
|
||||
// Best-effort only.
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user