feat(mqtt): MqttConnection connect/TLS/auth under bounded deadline
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,326 @@
|
||||
using System.Net.Security;
|
||||
using System.Security.Cryptography.X509Certificates;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MQTTnet;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
|
||||
|
||||
/// <summary>
|
||||
/// Owns the driver's single live MQTTnet-5 client: assembles the broker connection options
|
||||
/// from <see cref="MqttDriverOptions"/> (endpoint, protocol version, session, credentials,
|
||||
/// TLS + CA pin) and connects under a bounded deadline.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Connection-free construction.</b> The constructor stores configuration only — it
|
||||
/// opens no socket and creates no client. The universal-browser <c>CanBrowse</c> pattern
|
||||
/// constructs a throwaway instance purely to ask what driver type it is, so a
|
||||
/// constructor that dialled a broker would stall the AdminUI.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Bounded connect.</b> <see cref="ConnectAsync"/> links the caller's token with a
|
||||
/// <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/> deadline, and the same value is
|
||||
/// fed to MQTTnet's own <see cref="MqttClientOptions.Timeout"/>. 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Credentials never leak.</b> Nothing here logs or formats
|
||||
/// <see cref="MqttDriverOptions.Password"/>; the options record itself redacts it in
|
||||
/// <c>ToString()</c>.
|
||||
/// </para>
|
||||
/// Reconnect/backoff (Task 4) and subscription (Task 6) are deliberately not implemented here.
|
||||
/// </remarks>
|
||||
public sealed class MqttConnection : IAsyncDisposable
|
||||
{
|
||||
private readonly string _driverId;
|
||||
private readonly ILogger? _logger;
|
||||
private readonly MqttDriverOptions _options;
|
||||
|
||||
private IMqttClient? _client;
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>Stores configuration only — no network, no client instantiation.</summary>
|
||||
/// <param name="options">Broker connection settings.</param>
|
||||
/// <param name="driverId">Driver instance id, used only for log/diagnostic correlation.</param>
|
||||
/// <param name="logger">Optional logger; never receives credentials.</param>
|
||||
public MqttConnection(MqttDriverOptions options, string driverId, ILogger? logger = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(options);
|
||||
ArgumentException.ThrowIfNullOrWhiteSpace(driverId);
|
||||
|
||||
_options = options;
|
||||
_driverId = driverId;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <summary>Whether the underlying client currently holds an established MQTT session.</summary>
|
||||
public bool IsConnected => _client?.IsConnected ?? false;
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (_disposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_disposed = true;
|
||||
|
||||
var client = Interlocked.Exchange(ref _client, null);
|
||||
if (client is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Connects to the broker, failing at <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/>
|
||||
/// rather than waiting indefinitely on an unresponsive peer.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Caller cancellation; linked with the connect deadline.</param>
|
||||
/// <exception cref="TimeoutException">The connect deadline elapsed.</exception>
|
||||
/// <exception cref="OperationCanceledException"><paramref name="cancellationToken"/> was cancelled.</exception>
|
||||
public async Task ConnectAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
// Single-caller by contract: the reconnect supervisor (Task 4) owns connect/disconnect
|
||||
// sequencing, so no lock is taken here. The options are rebuilt per attempt so a rotated
|
||||
// CA file is picked up on the next connect rather than being pinned for the process life.
|
||||
var client = _client ??= new MqttClientFactory().CreateMqttClient();
|
||||
var clientOptions = BuildClientOptions(_options, clientIdSuffix: null, _logger);
|
||||
|
||||
using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds));
|
||||
using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, deadline.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 client.ConnectAsync(clientOptions, linked.Token).ConfigureAwait(false);
|
||||
}
|
||||
// MQTTnet wraps a cancelled connect in MqttConnectingFailedException rather than letting
|
||||
// the OperationCanceledException surface, so the two legs are told apart by which token
|
||||
// fired, not by the exception type. Caller intent wins over the deadline.
|
||||
catch (Exception ex) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw new OperationCanceledException(
|
||||
$"MQTT driver '{_driverId}': connect to {_options.Host}:{_options.Port} was cancelled.",
|
||||
ex,
|
||||
cancellationToken);
|
||||
}
|
||||
catch (Exception ex) when (deadline.IsCancellationRequested)
|
||||
{
|
||||
throw new TimeoutException(
|
||||
$"MQTT driver '{_driverId}': connect to {_options.Host}:{_options.Port} did not complete within "
|
||||
+ $"{_options.ConnectTimeoutSeconds}s.",
|
||||
ex);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Assembles the MQTTnet client options from <paramref name="options"/>. Pure — it performs
|
||||
/// no I/O and touches no network, so it is safe to call on a config-validation path. A
|
||||
/// configured <see cref="MqttDriverOptions.CaCertificatePath"/> is read lazily by the
|
||||
/// returned certificate-validation callback at handshake time, not here.
|
||||
/// </summary>
|
||||
/// <param name="options">Broker connection settings.</param>
|
||||
/// <param name="clientIdSuffix">
|
||||
/// Appended to <see cref="MqttDriverOptions.ClientId"/>, 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.
|
||||
/// </param>
|
||||
/// <param name="logger">Optional logger for the deferred CA-pin load; never receives credentials.</param>
|
||||
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));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Builds a certificate-validation callback that pins the broker chain to the PEM CA at
|
||||
/// <paramref name="caCertificatePath"/>. The file is loaded lazily on first validation
|
||||
/// (i.e. during the TLS handshake) so that this — and therefore
|
||||
/// <see cref="BuildClientOptions"/> — stays free of I/O. Any failure to load or to chain
|
||||
/// up to the pinned CA rejects the certificate: fail closed.
|
||||
/// </summary>
|
||||
private static Func<MqttClientCertificateValidationEventArgs, bool> CreatePinnedCaValidator(
|
||||
string caCertificatePath,
|
||||
ILogger? logger)
|
||||
{
|
||||
var trustedRoots = new Lazy<X509Certificate2Collection?>(
|
||||
() => 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);
|
||||
chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;
|
||||
|
||||
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;
|
||||
}
|
||||
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."),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net10.0</TargetFramework>
|
||||
<Nullable>enable</Nullable>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<LangVersion>latest</LangVersion>
|
||||
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
|
||||
<GenerateDocumentationFile>true</GenerateDocumentationFile>
|
||||
<NoWarn>$(NoWarn);CS1591</NoWarn>
|
||||
<RootNamespace>ZB.MOM.WW.OtOpcUa.Driver.Mqtt</RootNamespace>
|
||||
<AssemblyName>ZB.MOM.WW.OtOpcUa.Driver.Mqtt</AssemblyName>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj"/>
|
||||
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core.Abstractions\ZB.MOM.WW.OtOpcUa.Core.Abstractions.csproj"/>
|
||||
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core\ZB.MOM.WW.OtOpcUa.Core.csproj"/>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="MQTTnet"/>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<InternalsVisibleTo Include="ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests"/>
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
Reference in New Issue
Block a user