using System.Buffers;
using System.Net.Security;
using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
using Microsoft.Extensions.Logging;
using MQTTnet;
using MQTTnet.Protocol;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
///
/// Receives one inbound MQTT application message. Invoked on MQTTnet's own dispatcher
/// thread: an implementation must not block, must not do I/O and must not throw, or it stalls
/// the client's pump for every other subscription.
///
/// The concrete topic the message arrived on (never a filter).
///
/// The message body. Only valid for the duration of the call — the underlying buffer belongs to
/// MQTTnet, so anything an implementation wants to keep must be copied out.
///
///
/// The message's MQTT retain flag as delivered to this client, i.e. true for
/// the broker's stored last-known value replayed at subscribe time, and false for an
/// ordinary live publish. This is the retained seed signal.
///
public delegate void MqttMessageObserver(string topic, ReadOnlySpan payload, bool retained);
/// One topic filter to SUBSCRIBE, as the subscription manager wants it established.
/// The topic filter (concrete, or carrying MQTT wildcards).
/// Requested QoS, 0–2.
///
/// Whether the broker should replay its retained message for this filter at subscribe time.
/// Honoured natively on MQTT 5.0 (retain handling); on 3.1.1 the broker always replays and the
/// manager drops the seed client-side instead.
///
public sealed record MqttTopicSubscription(string Topic, int Qos, bool SeedRetained);
/// The broker's SUBACK verdict for one requested filter.
/// The filter this outcome answers.
/// Whether the broker granted the subscription.
/// The broker's reason code / string, for logs and diagnostics.
public sealed record MqttSubscribeOutcome(string Topic, bool Granted, string Reason);
///
/// The one seam through which the subscription manager establishes MQTT subscriptions.
/// is the production implementation; the interface exists so the
/// manager's SUBACK handling (per-filter grant / rejection, total failure) is exercisable without
/// a broker.
///
public interface IMqttSubscribeTransport
{
///
/// Issues one SUBSCRIBE carrying every filter and returns the broker's per-filter verdict.
/// Implementations must be bounded — a broker that accepts SUBSCRIBE and never SUBACKs must
/// fail at a deadline, not hang.
///
/// The filters to establish; never empty.
/// Caller cancellation, linked with the implementation's deadline.
/// One outcome per requested filter.
Task> SubscribeAsync(
IReadOnlyList filters,
CancellationToken cancellationToken);
}
/// Lifecycle state of an .
public enum MqttConnectionState
{
/// Constructed, or the first connect has not yet succeeded. No supervisor is running.
Disconnected = 0,
/// An MQTT session is established.
Connected = 1,
///
/// The session dropped and the supervisor is retrying under backoff. A broker that is merely
/// down stays here indefinitely — being unreachable is never a fault.
///
Reconnecting = 2,
///
/// Unrecoverable without a driver restart: retrying cannot help, so the supervisor has
/// stopped. Two causes — unusable configuration (a failure raised while assembling the client
/// options, before any I/O), or the supervisor itself dying on an unexpected exception. A
/// broker that is merely unreachable never lands here.
///
Faulted = 3,
/// has run. Terminal.
Disposed = 4,
}
///
/// The broker completed the handshake and refused the CONNECT, returning a non-success
/// CONNACK reason code.
///
///
///
/// This exists because MQTTnet 5 does not throw on an unsuccessful CONNACK — it returns
/// the outcome in and leaves the client
/// disconnected, and v5 removed v4's ThrowOnNonSuccessfulConnectResponse option, so
/// inspecting the result is the only way to see it. A caller that ignores the result gets a
/// connect that "succeeded" against a broker which rejected it: a driver configured with the
/// wrong broker password would report Healthy and never receive a single value. That is
/// exactly what the Task-13 live fixture caught.
///
///
/// distinguishes a rejection no amount of retrying can fix
/// (credentials, identity, protocol, an invalid Last-Will) from a genuinely transient refusal
/// (broker unavailable/busy, quota, rate limit). See
/// for the full mapping.
///
///
public sealed class MqttConnectRejectedException : Exception
{
internal MqttConnectRejectedException(MqttClientConnectResultCode resultCode, bool isUnrecoverable, string message)
: base(message)
{
ResultCode = resultCode;
IsUnrecoverable = isUnrecoverable;
}
/// Whether retrying against this broker with this configuration can never succeed.
public bool IsUnrecoverable { get; }
/// The CONNACK reason code the broker returned.
public MqttClientConnectResultCode ResultCode { get; }
}
///
/// Owns the driver's single live MQTTnet-5 client: assembles the broker connection options
/// from (endpoint, protocol version, session, credentials,
/// TLS + CA pin), connects under a bounded deadline, and keeps the session alive with a
/// hand-rolled reconnect supervisor.
///
///
///
/// Connection-free construction. The constructor stores configuration only — it
/// opens no socket, creates no client and starts no supervisor. The universal-browser
/// CanBrowse pattern constructs a throwaway instance purely to ask what driver type
/// it is, so a constructor that dialled a broker would stall the AdminUI.
///
///
/// Bounded connect. links the caller's token with a
/// deadline, and the same value is
/// fed to MQTTnet's own . A broker that accepts the
/// TCP connection and then never answers CONNACK — the frozen-peer shape that wedged the
/// S7 read path (arch-review R2-01) — fails at the deadline instead of hanging forever.
/// Every wait in this type is bounded the same way, teardown included.
///
///
/// Hand-rolled reconnect. MQTTnet v5 dropped v4's ManagedMqttClient, so there
/// is no library-managed reconnect: this type supplies it. The supervisor starts on the
/// first successful connect (a failed first attempt is the caller's to retry — the
/// driver-host resilience layer re-runs InitializeAsync — and must not leave a
/// background task hammering an endpoint the caller has given up on). MQTTnet's
/// DisconnectedAsync callback only signals a semaphore and returns, so the library's
/// own dispatcher thread is never blocked by a backoff sleep.
///
///
/// is load-bearing. MQTT subscriptions do not survive a
/// clean session, and a reconnect that completes without re-subscribing produces a
/// healthy-looking connection that receives nothing, forever, with no error, no exception
/// and no bad status. So the callback fires on every successful reconnect — including
/// persistent-session reconnects, where re-subscribing is merely redundant. Re-subscribing
/// an already-subscribed topic is harmless; missing one is silent death. If the callback
/// fails, the freshly-established session is torn down and retried rather than left
/// connected-but-deaf. It is invoked under a cancellation token and a
/// ceiling: a subscriber that hangs on
/// a broker which accepts SUBSCRIBE and never SUBACKs — the frozen-peer shape again — must
/// not be able to park the supervisor or outlive .
///
///
/// Connect is idempotent. MQTTnet throws (and raises no DisconnectedAsync) if
/// asked to connect a client that is already connected, so both connect paths check first and
/// return a no-op. The two paths genuinely race: the supervisor is told to expect the driver
/// host to re-run InitializeAsync, so a caller can restore the session while the
/// supervisor sits in backoff, and the supervisor can restore it a moment before a caller
/// asks. Without the check, the first case leaves the supervisor failing forever against a
/// perfectly healthy connection (inflating attempt, so the next genuine outage
/// waits instead of min) and the
/// second throws an undocumented out of
/// on a working session. When the supervisor finds the session
/// already restored it stops retrying without firing — the
/// caller that reconnected owns its own subscribe, per the
/// contract.
///
///
/// State transitions. is terminal; nothing
/// moves off it.
///
/// -
///
///
/// set by on success (or on finding the session
/// already up), and by the supervisor only after
/// has completed — so on the reconnect path
/// Connected means "connected and re-subscribed".
///
///
/// -
///
///
/// set by the DisconnectedAsync callback when a live session drops, and
/// held by the supervisor for the whole retry + re-subscribe sequence.
///
///
/// -
///
///
/// set when the client options cannot be assembled, or when the supervisor exits
/// on an unexpected exception. Both mean no further recovery without a restart.
///
///
/// -
///
/// set first thing in .
///
///
/// and can legitimately disagree in both
/// directions and are not interchangeable: Reconnecting with
/// IsConnected == true is the window where the socket is up but the re-subscribe has
/// not finished, and Connected with IsConnected == false is the instant between
/// the socket dropping and MQTTnet raising its callback. is the health
/// surface; is the transport fact.
///
///
/// Credentials never leak. Nothing here logs or formats
/// ; the options record itself redacts it in
/// ToString().
///
///
/// Lifecycle is serialised. Connect (caller-initiated or supervisor-initiated) and
/// dispose run under a single gate, and a connect that completes re-checks the disposed flag
/// while still holding that gate. This closes the connection leak the Task-3 review
/// traced: dispose could observe a still-null client, dispose nothing and return, after
/// which the in-flight connect assigned a client and connected successfully — leaving a live
/// socket no later dispose could ever reach. Now the losing side of that race disposes the
/// client it created and reports .
///
///
/// The publish leg is deliberately narrow. exists to satisfy
/// — whose entire production caller is
/// — not to make this a general MQTT publisher. This
/// driver has no IWritable leg in v1, so a Sparkplug rebirth NCMD is the only
/// outbound application message it ever sends. Sparkplug rebirth-on-reconnect itself is decided
/// by hanging off ; this type
/// only carries the bytes.
///
///
public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, Sparkplug.IMqttPublishTransport
{
private readonly string _driverId;
///
/// Serialises connect (both callers of it) against dispose. Never disposed: a late MQTTnet
/// callback can still reach this instance after teardown, and a disposed
/// would throw inside the library's own dispatcher.
///
private readonly SemaphoreSlim _lifecycleGate = new(1, 1);
///
/// Cancelled by ; aborts the supervisor, any in-flight connect and
/// any in-flight callback. Never disposed — deliberately, like the
/// two semaphores: a connect racing dispose reads
/// after its own disposed check, and disposing the source would turn that benign loser into an
/// naming the wrong type. It holds no timer and no
/// surviving registrations (every linked source here is using-scoped), so leaving it
/// undisposed costs nothing.
///
private readonly CancellationTokenSource _lifetimeCts = new();
private readonly ILogger? _logger;
private readonly MqttDriverOptions _options;
///
/// Wakes the supervisor. Unbounded max count: spurious releases (a failed connect attempt
/// also raises DisconnectedAsync) are drained harmlessly by the still-connected guard,
/// whereas a bounded semaphore would throw on the extra release. Never disposed, for the same
/// reason as .
///
private readonly SemaphoreSlim _reconnectWake = new(0);
private IMqttClient? _client;
private int _disposed;
private long _lastMessageTicksUtc;
private Task? _reconnectWorker;
private int _state = (int)MqttConnectionState.Disconnected;
/// Stores configuration only — no network, no client instantiation, no supervisor.
/// Broker connection settings.
/// Driver instance id, used only for log/diagnostic correlation.
/// Optional logger; never receives credentials.
public MqttConnection(MqttDriverOptions options, string driverId, ILogger? logger = null)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentException.ThrowIfNullOrWhiteSpace(driverId);
_options = options;
_driverId = driverId;
_logger = logger;
}
///
/// Raised after every successful reconnect (never after the initial
/// , which the caller already sequences its own subscribe behind,
/// and never when the supervisor merely finds the session already restored by such a caller).
/// Subscribers must re-establish their MQTT subscriptions here — see the type remarks for
/// why skipping it is silent death. Every subscriber is invoked even if an earlier one
/// throws; a throwing subscriber causes the session to be torn down and retried.
///
///
/// The supplied token is cancelled by , and the whole fan-out is
/// additionally capped at — a subscriber
/// that ignores its token cannot park the supervisor or make it outlive teardown. Subscribers
/// should flow the token into their own network calls rather than relying on that ceiling,
/// which abandons rather than stops the offending work.
///
public event Func? Reconnected;
///
/// Raised for every inbound application message, on MQTTnet's own dispatcher thread. This
/// connection does no routing, parsing or typing of its own — that is the subscription
/// manager's job; here the message is only stamped onto and
/// handed on.
///
///
/// Handlers must not block, do I/O or throw. A throwing handler is caught and logged here
/// rather than being allowed to escape into the library's pump — one bad tag must not stop
/// delivery for every other subscriber.
///
public event MqttMessageObserver? MessageReceived;
/// Whether the underlying client currently holds an established MQTT session.
public bool IsConnected => _client?.IsConnected ?? false;
///
/// UTC timestamp of the most recent inbound application message, or null if none has
/// arrived on this instance. Connection health only — nothing here routes or parses the
/// message; that is the subscription manager's job (Task 6).
///
///
/// The plan calls this member LastMessageAgeUtc, which conflates a timestamp with an
/// age; it is split here into the instant () and the elapsed span
/// ().
///
public DateTime? LastMessageUtc
{
get
{
var ticks = Interlocked.Read(ref _lastMessageTicksUtc);
return ticks == 0 ? null : new DateTime(ticks, DateTimeKind.Utc);
}
}
///
/// How long ago the most recent inbound application message arrived, or null if none
/// has. A connection that is with a large age is
/// the "reconnected but never re-subscribed" shape.
///
public TimeSpan? LastMessageAge => LastMessageUtc is { } at ? DateTime.UtcNow - at : null;
/// Current lifecycle state. See .
public MqttConnectionState State => (MqttConnectionState)Volatile.Read(ref _state);
///
/// Test seam: awaited inside the gated connect immediately after the broker session is
/// established and before the disposed re-check, so a test can reproduce the exact
/// connect/dispose interleaving that used to leak a live connection. Always null in
/// production.
///
internal Func? AfterConnectHookForTests { get; set; }
private bool Disposed => Volatile.Read(ref _disposed) != 0;
///
public async ValueTask DisposeAsync()
{
if (Interlocked.Exchange(ref _disposed, 1) != 0)
{
return;
}
SetState(MqttConnectionState.Disposed);
// Stop the supervisor and abort any in-flight connect FIRST, so the gate below is not held
// for a full connect deadline by work that is already pointless.
try
{
await _lifetimeCts.CancelAsync().ConfigureAwait(false);
}
catch (ObjectDisposedException)
{
// Nothing to cancel.
}
var worker = Volatile.Read(ref _reconnectWorker);
if (worker is not null)
{
try
{
await worker.WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds)).ConfigureAwait(false);
}
catch (Exception ex)
{
_logger?.LogDebug(ex, "MQTT driver '{DriverId}': reconnect supervisor did not stop cleanly.", _driverId);
}
}
// Bounded even here: a wedged connect must not turn dispose into a hang. If the gate cannot
// be taken we still claim the client — double-disposing an MqttClient is harmless, leaking a
// live one is not.
var gated = await _lifecycleGate
.WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds))
.ConfigureAwait(false);
if (!gated)
{
_logger?.LogWarning(
"MQTT driver '{DriverId}': lifecycle gate still held at dispose; tearing the client down anyway.",
_driverId);
}
try
{
var client = Interlocked.Exchange(ref _client, null);
if (client is not null)
{
await CloseAsync(client).ConfigureAwait(false);
}
}
finally
{
if (gated)
{
_lifecycleGate.Release();
}
}
}
///
/// The backoff a supervisor waits before reconnect attempt :
/// minSeconds × 2^attempt, saturated at . Pure — no
/// clock, no state, no jitter (deliberately: the schedule is asserted by exact equality, and
/// with one connection per driver instance there is no thundering herd to spread out).
///
///
/// Computed in rather than by shifting, because min << attempt
/// wraps — 1 << 32 is 1 — which would silently turn a long outage back
/// into a once-a-second hammering of a dead broker. A maxSeconds below
/// is misconfiguration; the floor wins, because the floor
/// is what stops the loop going hot.
///
/// Zero-based attempt index; negatives are treated as the first attempt.
/// Delay before the first attempt, and the floor for every later one.
/// Cap on the exponential growth.
public static TimeSpan NextBackoff(int attempt, int minSeconds, int maxSeconds)
{
var min = Math.Max(1d, minSeconds);
var max = Math.Max(min, maxSeconds);
if (attempt <= 0)
{
return TimeSpan.FromSeconds(min);
}
// Math.Pow saturates to +Infinity instead of wrapping, so a huge attempt count clamps to max.
var seconds = min * Math.Pow(2d, attempt);
return TimeSpan.FromSeconds(seconds >= max || double.IsNaN(seconds) ? max : seconds);
}
///
/// Connects to the broker, failing at
/// rather than waiting indefinitely on an unresponsive peer. On success the reconnect
/// supervisor takes over keeping the session alive.
///
///
///
/// May be retried on the same instance after a failed attempt — the underlying
/// is created once and reused across attempts. Safe to call
/// concurrently with : the two are serialised, and a connect
/// that loses the race disposes whatever it built and throws
/// .
///
///
/// Idempotent. Calling it on a session that is already established — including one
/// the reconnect supervisor restored a moment earlier — is a no-op that returns normally,
/// not the MQTTnet would raise for a
/// connect-while-connected. The caller still owns re-establishing its own subscriptions
/// after this returns; is not fired for this path.
///
///
/// Caller cancellation; linked with the connect deadline.
/// The connect deadline elapsed.
/// was cancelled.
///
/// The connection was disposed, either before the attempt started or while it was in flight.
///
///
/// The broker refused the CONNECT (non-success CONNACK — bad credentials, unsupported protocol
/// version, server unavailable …). MQTTnet reports this as a result, not an exception,
/// so this method inspects the CONNACK and raises it: a caller that treats "did not throw" as
/// "connected" would otherwise report a healthy session the broker never granted.
///
public async Task ConnectAsync(CancellationToken cancellationToken)
{
ObjectDisposedException.ThrowIf(Disposed, this);
await ConnectCoreAsync(supervisorAttempt: false, cancellationToken).ConfigureAwait(false);
}
///
/// Invokes every subscriber. Walks the invocation list explicitly:
/// awaiting a multicast delegate directly yields only the last subscriber's task and
/// abandons the earlier ones, so a single throwing subscriber would silently skip every
/// subscriber behind it — and a skipped re-subscribe is a topic that goes dark. Failures are
/// collected and rethrown after every subscriber has had its turn.
///
/// Passed to every subscriber; cancelled by .
internal async Task FireReconnectedAsync(CancellationToken cancellationToken)
{
var subscribers = Reconnected;
if (subscribers is null)
{
return;
}
List? failures = null;
foreach (var subscriber in subscribers.GetInvocationList().Cast>())
{
try
{
await subscriber(cancellationToken).ConfigureAwait(false);
}
catch (Exception ex)
{
(failures ??= []).Add(ex);
}
}
if (failures is not null)
{
throw new AggregateException(
$"MQTT driver '{_driverId}': {failures.Count} re-subscribe callback(s) failed after reconnect.",
failures);
}
}
///
/// The single gated connect both entry points funnel through.
///
///
/// true when the reconnect supervisor is calling. It suppresses publishing
/// on success, because on the reconnect path
/// "connected" is only true once has re-subscribed — publishing it
/// here would advertise a healthy connection during the window in which it has no
/// subscriptions at all. It also suppresses starting the supervisor (it is the supervisor).
///
/// Caller cancellation; linked with the connect deadline.
///
/// true if this call established the session; false if it found one already
/// established and did nothing. The supervisor uses false to stop retrying without
/// firing — whoever established that session owns its subscribe.
///
private async Task ConnectCoreAsync(bool supervisorAttempt, CancellationToken cancellationToken)
{
// The options are rebuilt per attempt so a rotated CA file is picked up on the next connect
// rather than being pinned for the life of the process. This is also the only step that can
// fail unrecoverably — it is pure config assembly, so retrying it can never help.
MqttClientOptions clientOptions;
try
{
clientOptions = BuildClientOptions(_options, clientIdSuffix: null, _logger);
}
catch (Exception ex)
{
SetState(MqttConnectionState.Faulted);
_logger?.LogError(
ex,
"MQTT driver '{DriverId}': broker configuration is unusable; no amount of retrying will help.",
_driverId);
throw;
}
using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds));
using var linked = CancellationTokenSource.CreateLinkedTokenSource(
cancellationToken,
deadline.Token,
_lifetimeCts.Token);
_logger?.LogDebug(
"MQTT driver '{DriverId}': connecting to {Host}:{Port} (tls={UseTls}, timeout={TimeoutSeconds}s).",
_driverId,
_options.Host,
_options.Port,
_options.UseTls,
_options.ConnectTimeoutSeconds);
try
{
await _lifecycleGate.WaitAsync(linked.Token).ConfigureAwait(false);
}
catch (Exception ex)
{
throw Classify(ex, cancellationToken, deadline);
}
try
{
// Dispose won the race outright: it has already flagged the instance and disposed
// whatever existed. Creating a client now would be creating one nobody can reach.
ObjectDisposedException.ThrowIf(Disposed, this);
var client = _client;
if (client is null)
{
client = new MqttClientFactory().CreateMqttClient();
AttachHandlers(client);
_client = client;
}
// Idempotence, and the whole point of the gate: MQTTnet throws
// "not allowed to connect with a server after the connection is established" for a
// connect-while-connected AND raises no DisconnectedAsync for it, so neither the retry
// loop nor Classify would ever recover. Both entry points race for real — see the type
// remarks — so whichever arrives second must find the session and leave it alone.
if (client.IsConnected)
{
SetState(MqttConnectionState.Connected);
StartSupervisorIfCallerConnect(supervisorAttempt);
_logger?.LogDebug(
"MQTT driver '{DriverId}': connect to {Host}:{Port} skipped — the session is already established.",
_driverId,
_options.Host,
_options.Port);
return false;
}
var connack = await client.ConnectAsync(clientOptions, linked.Token).ConfigureAwait(false);
// MQTTnet 5 does NOT throw for a broker that refused the CONNECT — it hands the reason
// back here and leaves the client disconnected. Skipping this check is how a wrong broker
// password produced a Healthy driver that never received a value (Task-13 live gate).
if (connack is not null && connack.ResultCode != MqttClientConnectResultCode.Success)
{
throw BuildRejection(connack, supervisorAttempt);
}
if (AfterConnectHookForTests is { } hook)
{
await hook().ConfigureAwait(false);
}
// The race the Task-3 review traced: dispose flagged the instance while this connect was
// in flight, then blocked on the very gate this call holds — so it disposed nothing and
// the session below would have been unreachable forever. Hand it back here instead.
if (Disposed)
{
Interlocked.CompareExchange(ref _client, null, client);
await CloseAsync(client).ConfigureAwait(false);
throw new ObjectDisposedException(
nameof(MqttConnection),
$"MQTT driver '{_driverId}': the connection was disposed while a connect to "
+ $"{_options.Host}:{_options.Port} was in flight; the established session was torn down.");
}
// On the supervisor path the session is up but has no subscriptions yet, so State stays
// Reconnecting until FireReconnectedAsync succeeds — see the state table in the remarks.
if (!supervisorAttempt)
{
SetState(MqttConnectionState.Connected);
}
StartSupervisorIfCallerConnect(supervisorAttempt);
return true;
}
catch (ObjectDisposedException)
{
throw;
}
// A refused CONNECT is already fully classified — it must not be re-wrapped as a timeout or a
// cancellation just because a token happened to fire while the CONNACK was being read.
catch (MqttConnectRejectedException)
{
throw;
}
catch (Exception ex)
{
throw Classify(ex, cancellationToken, deadline);
}
finally
{
_lifecycleGate.Release();
}
}
///
/// Turns a refused CONNACK into the thrown contract, setting
/// first when the refusal is unrecoverable — which is what stops the supervisor
/// ( returns on Faulted) from retrying every
/// backoff period against a broker that will never accept these credentials.
///
private MqttConnectRejectedException BuildRejection(MqttClientConnectResult connack, bool supervisorAttempt)
{
var unrecoverable = IsUnrecoverableConnackRejection(connack.ResultCode);
var message = DescribeConnackRejection($"{_options.Host}:{_options.Port}", connack.ResultCode);
if (unrecoverable)
{
SetState(MqttConnectionState.Faulted);
_logger?.LogError(
"MQTT driver '{DriverId}': {Message} Retrying cannot help; fix the configuration and redeploy.",
_driverId,
message);
}
else
{
// Retryable: leave the state alone — Disconnected before a first connect, Reconnecting
// under the supervisor — so the backoff loop keeps trying at its bounded rate.
_logger?.LogWarning(
"MQTT driver '{DriverId}': {Message} Treating as transient{Suffix}.",
_driverId,
message,
supervisorAttempt ? " and continuing to retry" : string.Empty);
}
return new MqttConnectRejectedException(connack.ResultCode, unrecoverable, message);
}
///
/// Whether a refused CONNACK can never be fixed by retrying, and so must fault the connection
/// rather than leave it looping.
///
///
///
/// Unrecoverable covers credentials/identity (BadUserNameOrPassword,
/// NotAuthorized, ClientIdentifierNotValid, Banned,
/// BadAuthenticationMethod), protocol mismatch (UnsupportedProtocolVersion,
/// MalformedPacket, ProtocolError, PacketTooLarge) and a CONNECT whose
/// Last-Will the broker will not accept (TopicNameInvalid,
/// PayloadFormatInvalid, RetainNotSupported, QoSNotSupported). Every
/// one is a property of the configuration we keep re-sending.
///
///
/// ServerMoved (0x9D) is unrecoverable and UseAnotherServer (0x9C) is not,
/// because MQTT 5 defines the first as "permanently use another server" and the second as
/// "temporarily use another server" — the spec, not a guess.
///
///
/// Everything else — including UnspecifiedError, ImplementationSpecificError
/// and any code a future broker invents — is treated as transient. That default is
/// deliberate: wrongly faulting a transient refusal takes a recoverable driver down until
/// someone redeploys, whereas wrongly retrying a permanent one costs a bounded, visibly
/// Reconnecting, backoff-paced attempt that names the reason code in the log.
///
///
internal static bool IsUnrecoverableConnackRejection(MqttClientConnectResultCode code) => code switch
{
MqttClientConnectResultCode.MalformedPacket
or MqttClientConnectResultCode.ProtocolError
or MqttClientConnectResultCode.UnsupportedProtocolVersion
or MqttClientConnectResultCode.ClientIdentifierNotValid
or MqttClientConnectResultCode.BadUserNameOrPassword
or MqttClientConnectResultCode.NotAuthorized
or MqttClientConnectResultCode.Banned
or MqttClientConnectResultCode.BadAuthenticationMethod
or MqttClientConnectResultCode.TopicNameInvalid
or MqttClientConnectResultCode.PacketTooLarge
or MqttClientConnectResultCode.PayloadFormatInvalid
or MqttClientConnectResultCode.RetainNotSupported
or MqttClientConnectResultCode.QoSNotSupported
or MqttClientConnectResultCode.ServerMoved => true,
_ => false,
};
///
/// Renders a refused CONNACK as a targeted, credential-free operator message. Shared with
/// MqttDriverProbe so the AdminUI "Test connect" button and the running driver describe
/// the same rejection in the same words.
///
internal static string DescribeConnackRejection(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}.",
};
///
/// Starts the supervisor on the first successful caller connect. Always called while
/// holding the lifecycle gate, so the ??= needs no further synchronisation.
///
private void StartSupervisorIfCallerConnect(bool supervisorAttempt)
{
if (!supervisorAttempt)
{
_reconnectWorker ??= Task.Run(() => ReconnectSupervisorAsync(_lifetimeCts.Token));
}
}
///
/// Maps a bounded-operation failure onto this type's documented contract. MQTTnet wraps a
/// cancelled connect in MqttConnectingFailedException rather than letting the
/// surface, so the legs are told apart by which
/// token fired, not by exception type. Precedence: teardown, then caller intent, then the
/// deadline.
///
/// The failure to classify.
/// The caller's token — cancelled means caller intent.
/// The operation's own deadline source.
/// The operation name for the message ("connect", "subscribe").
private Exception Classify(
Exception ex,
CancellationToken cancellationToken,
CancellationTokenSource deadline,
string operation = "connect")
{
if (Disposed)
{
return new ObjectDisposedException(
nameof(MqttConnection),
new InvalidOperationException(
$"MQTT driver '{_driverId}': {operation} to {_options.Host}:{_options.Port} was aborted because the "
+ "connection was disposed while the attempt was in flight.",
ex));
}
if (cancellationToken.IsCancellationRequested)
{
return new OperationCanceledException(
$"MQTT driver '{_driverId}': {operation} to {_options.Host}:{_options.Port} was cancelled.",
ex,
cancellationToken);
}
if (deadline.IsCancellationRequested)
{
return new TimeoutException(
$"MQTT driver '{_driverId}': {operation} to {_options.Host}:{_options.Port} did not complete within "
+ $"{_options.ConnectTimeoutSeconds}s.",
ex);
}
return ex;
}
private void AttachHandlers(IMqttClient client)
{
client.DisconnectedAsync += OnDisconnectedAsync;
client.ApplicationMessageReceivedAsync += OnApplicationMessageReceivedAsync;
}
///
/// Runs on MQTTnet's own dispatcher, so it does exactly two non-blocking things and returns.
/// Sleeping the backoff here would stall the client's internal pump.
///
private Task OnDisconnectedAsync(MqttClientDisconnectedEventArgs args)
{
if (Disposed || State is MqttConnectionState.Faulted or MqttConnectionState.Disposed)
{
return Task.CompletedTask;
}
if (State == MqttConnectionState.Connected)
{
SetState(MqttConnectionState.Reconnecting);
}
// Released unconditionally rather than only when the client had been connected: a missed
// wake is a permanently dark connection, whereas a spurious one costs a loop iteration.
_reconnectWake.Release();
return Task.CompletedTask;
}
private Task OnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs args)
{
Interlocked.Exchange(ref _lastMessageTicksUtc, DateTime.UtcNow.Ticks);
var observers = MessageReceived;
if (observers is null)
{
return Task.CompletedTask;
}
var message = args.ApplicationMessage;
var payload = message.Payload;
// Same idiom the browse session uses: the common single-segment case is served straight off
// the library's buffer; a fragmented sequence is flattened once. Either way the span is only
// valid for this call, which is exactly what MqttMessageObserver documents.
ReadOnlySpan body = payload.IsSingleSegment ? payload.FirstSpan : payload.ToArray();
try
{
observers(message.Topic, body, message.Retain);
}
catch (Exception ex)
{
// An exception escaping here surfaces inside MQTTnet's own pump. Contain it: a broken
// observer must degrade its own tags, never stop delivery for every other subscription.
_logger?.LogError(
ex,
"MQTT driver '{DriverId}': an inbound-message observer threw; the message was dropped.",
_driverId);
}
return Task.CompletedTask;
}
///
/// Issues one SUBSCRIBE carrying every requested filter, bounded by
/// so a broker that accepts SUBSCRIBE and
/// never answers SUBACK — the frozen-peer shape again — fails at the deadline instead of
/// parking the caller (and, on the reconnect path, the supervisor) forever.
///
///
/// The deadline is this method's own linked source, deliberately not
/// : that knob already governs every MQTTnet operation
/// and repurposing it would couple the subscribe budget to the connect budget in a way neither
/// side could change independently.
///
/// A rejected filter is not an exception: the returned outcomes carry the broker's
/// per-filter verdict so the caller can degrade exactly the affected references. Only a
/// failure of the SUBSCRIBE itself (transport error, deadline, teardown) throws.
///
///
/// The filters to establish.
/// Caller cancellation; linked with the subscribe deadline.
/// One outcome per requested filter, in request order.
/// The subscribe deadline elapsed.
/// was cancelled.
/// The connection was disposed.
/// There is no established session to subscribe on.
public async Task> SubscribeAsync(
IReadOnlyList filters,
CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(filters);
ObjectDisposedException.ThrowIf(Disposed, this);
if (filters.Count == 0)
{
return [];
}
var client = _client
?? throw new InvalidOperationException(
$"MQTT driver '{_driverId}': cannot subscribe before a session to {_options.Host}:{_options.Port} "
+ "has been established.");
var builder = new MqttClientSubscribeOptionsBuilder();
foreach (var filter in filters)
{
builder = builder.WithTopicFilter(f => f
.WithTopic(filter.Topic)
.WithQualityOfServiceLevel(MapQos(filter.Qos))
// Only meaningful on MQTT 5.0. On 3.1.1 the broker always replays its retained
// message, so the manager also drops unwanted seeds client-side off the retain flag.
.WithRetainHandling(filter.SeedRetained
? MqttRetainHandling.SendAtSubscribe
: MqttRetainHandling.DoNotSendOnSubscribe));
}
using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds));
using var linked = CancellationTokenSource.CreateLinkedTokenSource(
cancellationToken,
deadline.Token,
_lifetimeCts.Token);
MqttClientSubscribeResult result;
try
{
result = await client.SubscribeAsync(builder.Build(), linked.Token).ConfigureAwait(false);
}
catch (Exception ex)
{
throw Classify(ex, cancellationToken, deadline, operation: "subscribe");
}
// Pair each requested filter with its SUBACK item positionally: MQTT guarantees SUBACK reason
// codes arrive in the order of the SUBSCRIBE's filters. A short/absent SUBACK (a
// specification-violating broker) is reported as ungranted rather than silently assumed good.
var items = result.Items as IList ?? [.. result.Items];
var outcomes = new List(filters.Count);
for (var i = 0; i < filters.Count; i++)
{
if (i >= items.Count)
{
outcomes.Add(new MqttSubscribeOutcome(filters[i].Topic, Granted: false, "NoSubAckReasonCode"));
continue;
}
var code = items[i].ResultCode;
var granted = code is MqttClientSubscribeResultCode.GrantedQoS0
or MqttClientSubscribeResultCode.GrantedQoS1
or MqttClientSubscribeResultCode.GrantedQoS2;
outcomes.Add(new MqttSubscribeOutcome(filters[i].Topic, granted, code.ToString()));
}
return outcomes;
}
///
/// Publishes one application message, bounded by
/// so a broker that accepts PUBLISH and
/// never completes it cannot park the caller — the same frozen-peer rule every other wait here
/// follows.
///
///
///
/// A refused PUBLISH throws. MQTTnet reports a broker rejection as a
/// result (as it does for CONNACK — see ),
/// so a caller that treated "did not throw" as "published" would silently drop a rebirth
/// NCMD against a broker whose ACL denies the NCMD topic and then wait forever for the
/// birth it never asked for.
///
///
/// Publishing on a down session is refused up front rather than left to MQTTnet's
/// own exception, so the message names the driver and the broker. A rebirth request lost to
/// a reconnect is not a fault: the reconnect path re-requests one itself.
///
///
/// The concrete topic to publish on.
/// The message body.
/// Requested QoS, 0–2; out-of-range values clamp.
/// The MQTT retain flag.
/// Caller cancellation; linked with the publish deadline.
/// A task that completes when the broker has accepted the message.
/// The publish deadline elapsed.
/// was cancelled.
/// The connection was disposed.
/// There is no established session, or the broker refused the message.
public async Task PublishAsync(
string topic,
byte[] payload,
int qos,
bool retain,
CancellationToken cancellationToken)
{
ArgumentException.ThrowIfNullOrEmpty(topic);
ArgumentNullException.ThrowIfNull(payload);
ObjectDisposedException.ThrowIf(Disposed, this);
var client = _client;
if (client is null || !client.IsConnected)
{
throw new InvalidOperationException(
$"MQTT driver '{_driverId}': cannot publish '{topic}' — there is no established session to "
+ $"{_options.Host}:{_options.Port}.");
}
var message = new MqttApplicationMessageBuilder()
.WithTopic(topic)
.WithPayload(payload)
.WithQualityOfServiceLevel(MapQos(qos))
.WithRetainFlag(retain)
.Build();
using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds));
using var linked = CancellationTokenSource.CreateLinkedTokenSource(
cancellationToken,
deadline.Token,
_lifetimeCts.Token);
MqttClientPublishResult result;
try
{
result = await client.PublishAsync(message, linked.Token).ConfigureAwait(false);
}
catch (Exception ex)
{
throw Classify(ex, cancellationToken, deadline, operation: "publish");
}
if (result is not null && !result.IsSuccess)
{
throw new InvalidOperationException(
$"MQTT driver '{_driverId}': broker at {_options.Host}:{_options.Port} refused PUBLISH "
+ $"'{topic}': {result.ReasonCode}.");
}
}
/// Maps a configured QoS integer onto MQTTnet's enum, clamping an out-of-range value.
private static MqttQualityOfServiceLevel MapQos(int qos) => qos switch
{
<= 0 => MqttQualityOfServiceLevel.AtMostOnce,
1 => MqttQualityOfServiceLevel.AtLeastOnce,
_ => MqttQualityOfServiceLevel.ExactlyOnce,
};
///
/// The reconnect loop MQTTnet v5 no longer provides. Started on the first successful connect
/// and stopped by cancelling the lifetime token; it owns every
/// reconnect attempt so the library's dispatcher never waits on one.
///
private async Task ReconnectSupervisorAsync(CancellationToken cancellationToken)
{
try
{
while (!cancellationToken.IsCancellationRequested)
{
await _reconnectWake.WaitAsync(cancellationToken).ConfigureAwait(false);
if (Disposed || State == MqttConnectionState.Faulted)
{
return;
}
if (IsConnected)
{
continue; // Spurious wake (e.g. a failed attempt's own disconnect callback).
}
SetState(MqttConnectionState.Reconnecting);
for (var attempt = 0; !cancellationToken.IsCancellationRequested && !Disposed; attempt++)
{
// A caller's ConnectAsync may have restored the session while we slept. Bail
// before burning the backoff, not just before burning an attempt.
if (IsConnected)
{
AdoptSessionRestoredByCaller();
break;
}
await Task
.Delay(
NextBackoff(
attempt,
_options.ReconnectMinBackoffSeconds,
_options.ReconnectMaxBackoffSeconds),
cancellationToken)
.ConfigureAwait(false);
bool established;
try
{
established = await ConnectCoreAsync(supervisorAttempt: true, cancellationToken)
.ConfigureAwait(false);
}
catch (ObjectDisposedException) when (Disposed)
{
return;
}
catch (Exception ex)
{
if (State == MqttConnectionState.Faulted)
{
_logger?.LogError(
ex,
"MQTT driver '{DriverId}': reconnect abandoned — the broker configuration is unusable.",
_driverId);
return;
}
_logger?.LogDebug(
ex,
"MQTT driver '{DriverId}': reconnect attempt {Attempt} to {Host}:{Port} failed.",
_driverId,
attempt + 1,
_options.Host,
_options.Port);
continue;
}
if (!established)
{
// Someone else restored the session between the backoff and the connect.
// Firing Reconnected here would re-subscribe on top of a caller that is about
// to do exactly that itself, and — worse — leaving the loop running would
// keep failing against a healthy connection forever.
AdoptSessionRestoredByCaller();
break;
}
if (!await TryReSubscribeAsync(cancellationToken).ConfigureAwait(false))
{
continue;
}
// Only now is "Connected" true in the sense this type sells it: connected AND
// subscribed.
SetState(MqttConnectionState.Connected);
_logger?.LogInformation(
"MQTT driver '{DriverId}': reconnected to {Host}:{Port} after {Attempts} attempt(s); "
+ "subscriptions re-established.",
_driverId,
_options.Host,
_options.Port,
attempt + 1);
break;
}
}
}
catch (OperationCanceledException)
{
// Dispose cancelled the lifetime token — the expected way this loop ends.
}
catch (Exception ex)
{
// Reaching here means the connection is permanently dark, which is exactly the failure
// this supervisor exists to prevent — it must never be swallowed quietly, and State must
// not keep reporting Reconnecting, which every consumer reads as "recovering".
SetState(MqttConnectionState.Faulted);
_logger?.LogError(
ex,
"MQTT driver '{DriverId}': reconnect supervisor stopped unexpectedly; the broker session will not "
+ "recover without a driver restart.",
_driverId);
}
}
///
/// The session came back without this supervisor's help — a caller's
/// won the race while we were in backoff. Publish the truth and stand down; deliberately does
/// not fire , because that caller owns its own re-subscribe.
///
private void AdoptSessionRestoredByCaller()
{
SetState(MqttConnectionState.Connected);
_logger?.LogDebug(
"MQTT driver '{DriverId}': reconnect stood down — the session to {Host}:{Port} was restored elsewhere.",
_driverId,
_options.Host,
_options.Port);
}
///
/// Fires — ALWAYS, on every reconnect, including persistent
/// sessions where it is merely redundant, because subscriptions do not survive a clean
/// session and a reconnect that skips this leaves a healthy-looking, permanently deaf client.
///
///
/// false if the re-subscribe failed or overran its deadline, having torn the session
/// down so the next attempt starts from a clean CONNECT — serving a connection that receives
/// nothing is the one outcome this type must never produce.
///
private async Task TryReSubscribeAsync(CancellationToken cancellationToken)
{
// Bounded like every other wait here. A subscriber that flows the token cancels promptly; one
// that ignores it is abandoned at the ceiling rather than being allowed to park the
// supervisor past DisposeAsync.
var fanOut = FireReconnectedAsync(cancellationToken);
// The ceiling abandons the task rather than stopping it, so its eventual failure must still
// be observed or it resurfaces as an unobserved TaskException on the finalizer thread.
_ = fanOut.ContinueWith(
static abandoned => _ = abandoned.Exception,
CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
try
{
await fanOut
.WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds), cancellationToken)
.ConfigureAwait(false);
return true;
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw; // Teardown — let the supervisor's own cancellation handling end the loop.
}
catch (Exception ex)
{
_logger?.LogError(
ex,
"MQTT driver '{DriverId}': re-subscribe after reconnect failed or overran {TimeoutSeconds}s; tearing "
+ "the session down and retrying rather than serving a connection that receives nothing.",
_driverId,
_options.ConnectTimeoutSeconds);
SetState(MqttConnectionState.Reconnecting);
await ForceDisconnectAsync().ConfigureAwait(false);
return false;
}
}
///
/// Drops a session that is connected but unusable, so the next supervisor pass starts from a
/// clean CONNECT. Best-effort and bounded — failure here just means the next attempt sees a
/// client that is already down.
///
private async Task ForceDisconnectAsync()
{
var client = _client;
if (client is null)
{
return;
}
try
{
using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds));
await client.DisconnectAsync(new MqttClientDisconnectOptions(), deadline.Token).ConfigureAwait(false);
}
catch (Exception ex)
{
_logger?.LogDebug(ex, "MQTT driver '{DriverId}': forced disconnect failed.", _driverId);
}
}
private async Task CloseAsync(IMqttClient client)
{
try
{
if (client.IsConnected)
{
// Bounded even on teardown — a wedged broker must not stall driver shutdown.
using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds));
await client.DisconnectAsync(new MqttClientDisconnectOptions(), deadline.Token).ConfigureAwait(false);
}
}
catch (Exception ex)
{
_logger?.LogDebug(ex, "MQTT driver '{DriverId}': disconnect during dispose failed; disposing anyway.", _driverId);
}
finally
{
client.Dispose();
}
}
///
/// is terminal — nothing may move off it. A
/// check-then-write would let a thread preempted between the two steps resurrect a disposed
/// connection's state, so the transition is a compare-and-swap: the terminal check and the
/// write are the same atomic operation.
///
private void SetState(MqttConnectionState state)
{
var desired = (int)state;
while (true)
{
var current = Volatile.Read(ref _state);
if (current == (int)MqttConnectionState.Disposed || current == desired)
{
return;
}
if (Interlocked.CompareExchange(ref _state, desired, current) == current)
{
return;
}
}
}
///
/// Assembles the MQTTnet client options from . Pure — it performs
/// no I/O and touches no network, so it is safe to call on a config-validation path. A
/// configured is read lazily by the
/// returned certificate-validation callback at handshake time, not here.
///
/// Broker connection settings.
///
/// Appended to , so a browse/probe session can share
/// the configured identity without colliding with the driver's own client id. When both are
/// empty MQTTnet generates a random client id.
///
/// Optional logger for the deferred CA-pin load; never receives credentials.
public static MqttClientOptions BuildClientOptions(
MqttDriverOptions options,
string? clientIdSuffix,
ILogger? logger = null)
{
ArgumentNullException.ThrowIfNull(options);
var builder = new MqttClientOptionsBuilder()
.WithTcpServer(options.Host, options.Port)
.WithProtocolVersion(MapProtocolVersion(options.ProtocolVersion))
.WithCleanSession(options.CleanSession)
.WithKeepAlivePeriod(TimeSpan.FromSeconds(options.KeepAliveSeconds))
.WithTimeout(TimeSpan.FromSeconds(options.ConnectTimeoutSeconds));
var clientId = string.Concat(options.ClientId ?? string.Empty, clientIdSuffix ?? string.Empty);
if (!string.IsNullOrWhiteSpace(clientId))
{
builder = builder.WithClientId(clientId);
}
// An empty username means "connect anonymously"; MQTTnet would otherwise send an empty
// CONNECT username, which some brokers treat as a failed auth rather than as anonymous.
if (!string.IsNullOrEmpty(options.Username))
{
builder = builder.WithCredentials(options.Username, options.Password ?? string.Empty);
}
builder = builder.WithTlsOptions(tls => ConfigureTls(tls, options, logger));
return builder.Build();
}
private static void ConfigureTls(MqttClientTlsOptionsBuilder tls, MqttDriverOptions options, ILogger? logger)
{
if (!options.UseTls)
{
tls.UseTls(false);
return;
}
tls.UseTls(true)
// Explicit SNI / hostname-verification target. Left unset, name validation depends on
// how MQTTnet infers the host from the endpoint — pin it to the configured host.
.WithTargetHost(options.Host);
if (options.AllowUntrustedServerCertificate)
{
// Dev / on-prem escape hatch, off unless explicitly enabled — mirrors the
// ServerHistorian TLS knobs. This is the ONLY branch that accepts a bad certificate.
tls.WithAllowUntrustedCertificates(true)
.WithIgnoreCertificateChainErrors(true)
.WithCertificateValidationHandler(static _ => true);
return;
}
if (string.IsNullOrWhiteSpace(options.CaCertificatePath))
{
// No pin: MQTTnet's default handler validates against the OS trust store.
return;
}
tls.WithCertificateValidationHandler(CreatePinnedCaValidator(options.CaCertificatePath, logger));
}
///
/// Builds a certificate-validation callback that pins the broker chain to the PEM CA at
/// . The file is loaded lazily on first validation
/// (i.e. during the TLS handshake) so that this — and therefore
/// — stays free of I/O. Any failure to load or to chain
/// up to the pinned CA rejects the certificate: fail closed.
///
private static Func CreatePinnedCaValidator(
string caCertificatePath,
ILogger? logger)
{
var trustedRoots = new Lazy(
() => LoadPemCa(caCertificatePath, logger),
LazyThreadSafetyMode.ExecutionAndPublication);
return args => ValidateAgainstPinnedCa(args, trustedRoots.Value, caCertificatePath, logger);
}
private static X509Certificate2Collection? LoadPemCa(string caCertificatePath, ILogger? logger)
{
try
{
var collection = new X509Certificate2Collection();
collection.ImportFromPemFile(caCertificatePath);
if (collection.Count != 0)
{
return collection;
}
logger?.LogError("MQTT CA pin '{CaCertificatePath}' contains no certificates; broker TLS will be rejected.", caCertificatePath);
}
catch (Exception ex)
{
logger?.LogError(ex, "MQTT CA pin '{CaCertificatePath}' could not be loaded; broker TLS will be rejected.", caCertificatePath);
}
return null;
}
private static bool ValidateAgainstPinnedCa(
MqttClientCertificateValidationEventArgs args,
X509Certificate2Collection? trustedRoots,
string caCertificatePath,
ILogger? logger)
{
if (trustedRoots is null || trustedRoots.Count == 0)
{
// Unreadable pin ⇒ no trust anchor ⇒ refuse. Never silently degrade to the OS store.
return false;
}
// The pin replaces the trust anchor only. Anything else the platform flagged — a hostname
// mismatch, an absent certificate — remains fatal.
if ((args.SslPolicyErrors & ~SslPolicyErrors.RemoteCertificateChainErrors) != SslPolicyErrors.None)
{
logger?.LogError("MQTT broker certificate rejected: {SslPolicyErrors}.", args.SslPolicyErrors);
return false;
}
if (args.Certificate is null)
{
return false;
}
var presented = args.Certificate as X509Certificate2;
X509Certificate2? owned = null;
if (presented is null)
{
owned = X509CertificateLoader.LoadCertificate(args.Certificate.Export(X509ContentType.Cert));
presented = owned;
}
try
{
using var chain = new X509Chain();
chain.ChainPolicy.TrustMode = X509ChainTrustMode.CustomRootTrust;
chain.ChainPolicy.CustomTrustStore.AddRange(trustedRoots);
// Revocation is deliberately not checked: the pin exists precisely for self-issued
// dev / on-prem CAs, which typically publish no CRL or OCSP responder, so checking
// would fail every such chain. The pin itself is the revocation story — remove the CA.
chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;
// Seed the intermediates the broker presented during the handshake. A leaf signed by
// an intermediate that chains up to the pinned root is a common broker topology, and
// that intermediate arrives on the wire rather than being installed locally — without
// it in the candidate pool a legitimate trust path simply cannot be built.
// CustomRootTrust still means only `trustedRoots` may terminate the chain, so a rogue
// self-signed cert smuggled in here cannot become an anchor.
// Both sources are read: platforms differ in whether the peer-supplied intermediates
// land in the incoming chain's elements, its ExtraStore, or both.
if (args.Chain is not null)
{
foreach (var element in args.Chain.ChainElements)
{
chain.ChainPolicy.ExtraStore.Add(element.Certificate);
}
chain.ChainPolicy.ExtraStore.AddRange(args.Chain.ChainPolicy.ExtraStore);
}
if (chain.Build(presented))
{
return true;
}
logger?.LogError(
"MQTT broker certificate does not chain to the pinned CA '{CaCertificatePath}': {ChainStatus}.",
caCertificatePath,
string.Join(", ", chain.ChainStatus.Select(s => s.Status)));
return false;
}
catch (CryptographicException ex)
{
// X509Chain.Build throws on a malformed certificate or an unusable policy. An exception
// escaping a TLS validation callback surfaces as an opaque handshake crash, so classify
// it here and refuse: an unverifiable certificate is a rejected certificate.
logger?.LogError(
ex,
"MQTT broker certificate could not be validated against the pinned CA '{CaCertificatePath}'; rejecting.",
caCertificatePath);
return false;
}
finally
{
owned?.Dispose();
}
}
private static MQTTnet.Formatter.MqttProtocolVersion MapProtocolVersion(MqttProtocolVersion version)
=> version switch
{
MqttProtocolVersion.V311 => MQTTnet.Formatter.MqttProtocolVersion.V311,
MqttProtocolVersion.V500 => MQTTnet.Formatter.MqttProtocolVersion.V500,
_ => throw new ArgumentOutOfRangeException(nameof(version), version, "Unsupported MQTT protocol version."),
};
}