diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs
index 54337abe..1a6e9072 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs
@@ -1,4 +1,5 @@
using System.Net.Security;
+using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
using Microsoft.Extensions.Logging;
using MQTTnet;
@@ -29,6 +30,18 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
/// ; the options record itself redacts it in
/// ToString().
///
+///
+/// NOT thread-safe — single-caller by contract. and
+/// take no lock and must never run concurrently. Calling them
+/// concurrently leaks a live broker connection: can
+/// observe _disposed == false, set it, exchange a still-null _client for
+/// null, and return having disposed nothing — after which the in-flight
+/// creates a client, assigns it and connects successfully.
+/// That client holds an open socket that no later can ever
+/// reach, because _disposed is already true. Serialising the lifecycle is
+/// the job of the reconnect supervisor added in Task 4; it is deliberately not solved
+/// here.
+///
/// Reconnect/backoff (Task 4) and subscription (Task 6) are deliberately not implemented here.
///
public sealed class MqttConnection : IAsyncDisposable
@@ -96,16 +109,24 @@ public sealed class MqttConnection : IAsyncDisposable
/// Connects to the broker, failing at
/// rather than waiting indefinitely on an unresponsive peer.
///
+ ///
+ /// May be retried on the same instance after a failed attempt — the underlying
+ /// is created once and reused across attempts, which is what the
+ /// Task-4 reconnect loop depends on. Not safe to call concurrently with itself or with
+ /// ; see the type-level remarks.
+ ///
/// 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.
+ ///
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.
+ // 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.
var client = _client ??= new MqttClientFactory().CreateMqttClient();
var clientOptions = BuildClientOptions(_options, clientIdSuffix: null, _logger);
@@ -124,6 +145,18 @@ public sealed class MqttConnection : IAsyncDisposable
{
await client.ConnectAsync(clientOptions, linked.Token).ConfigureAwait(false);
}
+ // A concurrent DisposeAsync disposes the client out from under the awaiting connect, and
+ // whatever MQTTnet throws for that matches neither token filter. Classify it rather than
+ // letting an undocumented exception escape the contract below.
+ catch (Exception ex) when (_disposed)
+ {
+ throw new ObjectDisposedException(
+ nameof(MqttConnection),
+ new InvalidOperationException(
+ $"MQTT driver '{_driverId}': connect to {_options.Host}:{_options.Port} was aborted because the "
+ + "connection was disposed while the attempt was in flight.",
+ ex));
+ }
// 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.
@@ -297,8 +330,30 @@ public sealed class MqttConnection : IAsyncDisposable
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;
@@ -310,6 +365,17 @@ public sealed class MqttConnection : IAsyncDisposable
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();
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs
index 45180483..c95cb714 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs
@@ -89,6 +89,45 @@ public sealed class MqttConnectionTests
await Should.ThrowAsync(async () => await conn.ConnectAsync(CancellationToken.None));
}
+ ///
+ /// Task 4's reconnect loop retries connect on the SAME instance, reusing the
+ /// IMqttClient held across failures. A failed attempt must therefore leave the
+ /// connection usable rather than poisoned.
+ ///
+ [Fact]
+ public async Task ConnectAsync_CanBeRetriedOnTheSameInstanceAfterAFailedAttempt()
+ {
+ using var blackhole = new BlackholeBroker();
+ var opts = new MqttDriverOptions
+ {
+ Host = "127.0.0.1", Port = blackhole.Port, UseTls = false, ConnectTimeoutSeconds = 1,
+ };
+ await using var conn = new MqttConnection(opts, driverId: "t", logger: null);
+
+ await Should.ThrowAsync(async () => await conn.ConnectAsync(CancellationToken.None));
+
+ // Second attempt reaches the broker again and fails the same way — not ObjectDisposed,
+ // not a null client, not a silently-swallowed no-op.
+ await Should.ThrowAsync(async () => await conn.ConnectAsync(CancellationToken.None));
+ conn.IsConnected.ShouldBeFalse();
+ }
+
+ ///
+ /// Connecting a disposed instance is an — the same
+ /// classification the in-flight dispose race folds into. Dispose is idempotent.
+ ///
+ [Fact]
+ public async Task ConnectAsync_AfterDispose_ThrowsObjectDisposed_AndDisposeIsIdempotent()
+ {
+ var opts = new MqttDriverOptions { Host = "127.0.0.1", Port = 1, UseTls = false, ConnectTimeoutSeconds = 1 };
+ var conn = new MqttConnection(opts, driverId: "t", logger: null);
+
+ await conn.DisposeAsync();
+ await conn.DisposeAsync();
+
+ await Should.ThrowAsync(async () => await conn.ConnectAsync(CancellationToken.None));
+ }
+
/// Never log or throw credentials — the plan's cross-cutting secrets rule.
[Fact]
public async Task ConnectAsync_Failure_DoesNotLeakPasswordIntoExceptionMessage()
@@ -153,21 +192,76 @@ public sealed class MqttConnectionTests
var tls = built.ChannelOptions.ShouldBeOfType().TlsOptions;
tls.CertificateValidationHandler.ShouldNotBeNull();
- tls.CertificateValidationHandler(SelfSignedValidationArgs(built)).ShouldBeFalse();
+ using var stranger = SelfSignedLeaf(Host);
+ tls.CertificateValidationHandler(ValidationArgs(built, stranger)).ShouldBeFalse();
}
/// An untrusted, unpinned server certificate is rejected — the pin actually validates.
[Fact]
public void BuildClientOptions_CaPin_RejectsCertificateNotIssuedByThePinnedCa()
{
- var caPath = WriteTempPemCa();
+ using var ca = IssueCa(issuer: null, "otopcua-test-ca", validDays: 30);
+ var caPath = WritePem(ca);
try
{
var opts = new MqttDriverOptions { Host = Host, Port = 8883, UseTls = true, CaCertificatePath = caPath };
var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
var tls = built.ChannelOptions.ShouldBeOfType().TlsOptions;
- tls.CertificateValidationHandler!(SelfSignedValidationArgs(built)).ShouldBeFalse();
+ using var stranger = SelfSignedLeaf(Host);
+ tls.CertificateValidationHandler!(ValidationArgs(built, stranger)).ShouldBeFalse();
+ }
+ finally
+ {
+ File.Delete(caPath);
+ }
+ }
+
+ ///
+ /// The accept path. Without this every certificate test asserts rejection, so deleting the
+ /// return true — or getting TrustMode / CustomTrustStore wrong — would
+ /// leave the suite green while no broker could ever connect.
+ ///
+ [Fact]
+ public void BuildClientOptions_CaPin_AcceptsLeafIssuedByThePinnedCa()
+ {
+ using var ca = IssueCa(issuer: null, "otopcua-test-ca", validDays: 30);
+ using var leaf = IssueLeaf(ca, Host, validDays: 20);
+ var caPath = WritePem(ca);
+ try
+ {
+ var opts = new MqttDriverOptions { Host = Host, Port = 8883, UseTls = true, CaCertificatePath = caPath };
+ var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
+ var tls = built.ChannelOptions.ShouldBeOfType().TlsOptions;
+
+ tls.CertificateValidationHandler!(ValidationArgs(built, leaf)).ShouldBeTrue();
+ }
+ finally
+ {
+ File.Delete(caPath);
+ }
+ }
+
+ ///
+ /// The common real-world broker topology: the leaf is signed by an intermediate that chains
+ /// to the pinned root, and the intermediate is delivered in the handshake rather than being
+ /// installed locally. Only the pinned ROOT is written to the CA file — the intermediate must
+ /// be picked up from what the broker presented.
+ ///
+ [Fact]
+ public void BuildClientOptions_CaPin_AcceptsLeafBehindAnIntermediateThatTheBrokerPresented()
+ {
+ using var root = IssueCa(issuer: null, "otopcua-test-root", validDays: 30);
+ using var intermediate = IssueCa(root, "otopcua-test-intermediate", validDays: 25);
+ using var leaf = IssueLeaf(intermediate, Host, validDays: 20);
+ var caPath = WritePem(root);
+ try
+ {
+ var opts = new MqttDriverOptions { Host = Host, Port = 8883, UseTls = true, CaCertificatePath = caPath };
+ var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
+ var tls = built.ChannelOptions.ShouldBeOfType().TlsOptions;
+
+ tls.CertificateValidationHandler!(ValidationArgs(built, leaf, intermediate)).ShouldBeTrue();
}
finally
{
@@ -290,29 +384,87 @@ public sealed class MqttConnectionTests
=> MqttConnection.BuildClientOptions(opts, clientIdSuffix: null)
.ChannelOptions.ShouldBeOfType().TlsOptions;
- private static MqttClientCertificateValidationEventArgs SelfSignedValidationArgs(MqttClientOptions built)
+ private static readonly DateTimeOffset NotBefore = DateTimeOffset.UtcNow.AddDays(-1);
+
+ /// A CA certificate, self-signed when is null.
+ private static X509Certificate2 IssueCa(X509Certificate2? issuer, string commonName, int validDays)
{
- using var rsa = RSA.Create(2048);
- var request = new CertificateRequest($"CN={Host}", rsa, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
- var cert = request.CreateSelfSigned(DateTimeOffset.UtcNow.AddDays(-1), DateTimeOffset.UtcNow.AddDays(1));
- return new MqttClientCertificateValidationEventArgs(
- cert,
- new X509Chain(),
- SslPolicyErrors.RemoteCertificateChainErrors,
- built.ChannelOptions);
+ using var key = RSA.Create(2048);
+ var request = new CertificateRequest($"CN={commonName}", key, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
+ request.CertificateExtensions.Add(new X509BasicConstraintsExtension(true, false, 0, true));
+ request.CertificateExtensions.Add(
+ new X509KeyUsageExtension(X509KeyUsageFlags.KeyCertSign | X509KeyUsageFlags.CrlSign, true));
+ request.CertificateExtensions.Add(new X509SubjectKeyIdentifierExtension(request.PublicKey, false));
+
+ if (issuer is null)
+ {
+ return request.CreateSelfSigned(NotBefore, NotBefore.AddDays(validDays));
+ }
+
+ // Create() drops the private key; an intermediate needs it back to sign its own children.
+ using var issued = request.Create(issuer, NotBefore, NotBefore.AddDays(validDays), NextSerial());
+ return issued.CopyWithPrivateKey(key);
}
- private static string WriteTempPemCa()
+ /// A server (end-entity) certificate signed by .
+ private static X509Certificate2 IssueLeaf(X509Certificate2 issuer, string commonName, int validDays)
+ {
+ using var key = RSA.Create(2048);
+ var request = new CertificateRequest($"CN={commonName}", key, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
+ request.CertificateExtensions.Add(new X509BasicConstraintsExtension(false, false, 0, true));
+ request.CertificateExtensions.Add(
+ new X509KeyUsageExtension(X509KeyUsageFlags.DigitalSignature | X509KeyUsageFlags.KeyEncipherment, true));
+ request.CertificateExtensions.Add(
+ new X509EnhancedKeyUsageExtension([new Oid("1.3.6.1.5.5.7.3.1")], false)); // serverAuth
+ request.CertificateExtensions.Add(new X509SubjectKeyIdentifierExtension(request.PublicKey, false));
+
+ return request.Create(issuer, NotBefore, NotBefore.AddDays(validDays), NextSerial());
+ }
+
+ /// A self-signed leaf that chains to nothing — the "stranger" a pin must reject.
+ private static X509Certificate2 SelfSignedLeaf(string commonName)
+ {
+ using var key = RSA.Create(2048);
+ var request = new CertificateRequest($"CN={commonName}", key, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
+ return request.CreateSelfSigned(NotBefore, NotBefore.AddDays(20));
+ }
+
+ private static byte[] NextSerial() => RandomNumberGenerator.GetBytes(8);
+
+ private static string WritePem(params X509Certificate2[] certificates)
{
- using var rsa = RSA.Create(2048);
- var request = new CertificateRequest("CN=otopcua-test-ca", rsa, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
- request.CertificateExtensions.Add(new X509BasicConstraintsExtension(true, false, 0, true));
- using var ca = request.CreateSelfSigned(DateTimeOffset.UtcNow.AddDays(-1), DateTimeOffset.UtcNow.AddDays(1));
var path = Path.Combine(Path.GetTempPath(), $"otopcua-mqtt-ca-{Guid.NewGuid():N}.pem");
- File.WriteAllText(path, ca.ExportCertificatePem());
+ File.WriteAllText(path, string.Join(Environment.NewLine, certificates.Select(c => c.ExportCertificatePem())));
return path;
}
+ ///
+ /// Mimics what SslStream hands a validation callback: the presented leaf plus a chain
+ /// carrying whatever intermediates the peer supplied. The chain is built (and allowed to
+ /// fail) purely so ChainElements is populated the way the platform would populate it.
+ ///
+ private static MqttClientCertificateValidationEventArgs ValidationArgs(
+ MqttClientOptions built,
+ X509Certificate2 leaf,
+ params X509Certificate2[] presentedIntermediates)
+ {
+ var chain = new X509Chain();
+ chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;
+ chain.ChainPolicy.VerificationFlags = X509VerificationFlags.AllFlags;
+ foreach (var intermediate in presentedIntermediates)
+ {
+ chain.ChainPolicy.ExtraStore.Add(intermediate);
+ }
+
+ chain.Build(leaf);
+
+ return new MqttClientCertificateValidationEventArgs(
+ leaf,
+ chain,
+ SslPolicyErrors.RemoteCertificateChainErrors,
+ built.ChannelOptions);
+ }
+
///
/// Accepts the TCP connection and then says nothing — the client sends CONNECT and waits
/// forever for a CONNACK that never comes. This is the frozen-peer shape; a closed port is