fix(mqtt): pin the CA accept path, honour presented intermediates, classify the dispose race

Review follow-ups on MqttConnection (Task 3):

- Every certificate test asserted rejection, so the accept branch was
  unreachable-by-regression. Adds a leaf genuinely issued by the pinned CA and
  asserts acceptance.
- ValidateAgainstPinnedCa never seeded ChainPolicy.ExtraStore from the incoming
  chain, so a leaf behind an intermediate delivered during the handshake failed
  despite a legitimate path to the pinned root. Seeds from both the incoming
  chain's elements and its ExtraStore; CustomRootTrust still means only the
  pinned roots may terminate the chain.
- A DisposeAsync racing an in-flight connect escaped as an unclassified
  exception; it now folds into ObjectDisposedException.
- Promotes the single-caller concurrency invariant into the type remarks, with
  the accurate blast radius (a leaked live connection, not a benign throw).
  Serialising the lifecycle remains Task 4's job.
- X509Chain.Build can throw; an exception escaping a TLS validation callback is
  an opaque handshake crash, so it is caught and refused.
- Adds a connect-retry test (Task 4's reconnect loop reuses the instance) and a
  disposed-then-connect test.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW

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;
 ///         <see cref="MqttDriverOptions.Password"/>; the options record itself redacts it in
 ///         <c>ToString()</c>.
 ///     </para>
+///     <para>
+///         <b>NOT thread-safe — single-caller by contract.</b> <see cref="ConnectAsync"/> and
+///         <see cref="DisposeAsync"/> take no lock and must never run concurrently. Calling them
+///         concurrently <b>leaks a live broker connection</b>: <see cref="DisposeAsync"/> can
+///         observe <c>_disposed == false</c>, set it, exchange a still-null <c>_client</c> for
+///         null, and return having disposed nothing — after which the in-flight
+///         <see cref="ConnectAsync"/> creates a client, assigns it and connects successfully.
+///         That client holds an open socket that no later <see cref="DisposeAsync"/> can ever
+///         reach, because <c>_disposed</c> is already <c>true</c>. Serialising the lifecycle is
+///         the job of the reconnect supervisor added in Task 4; it is deliberately not solved
+///         here.
+///     </para>
 ///     Reconnect/backoff (Task 4) and subscription (Task 6) are deliberately not implemented here.
 /// </remarks>
 public sealed class MqttConnection : IAsyncDisposable
@@ -96,16 +109,24 @@ public sealed class MqttConnection : IAsyncDisposable
     ///     Connects to the broker, failing at <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/>
     ///     rather than waiting indefinitely on an unresponsive peer.
     /// </summary>
+    /// <remarks>
+    ///     May be retried on the same instance after a failed attempt — the underlying
+    ///     <see cref="IMqttClient"/> 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 cref="DisposeAsync"/>; see the type-level remarks.
+    /// </remarks>
     /// <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>
+    /// <exception cref="ObjectDisposedException">
+    ///     The connection was disposed, either before the attempt started or while it was in flight.
+    /// </exception>
     public async Task ConnectAsync(CancellationToken cancellationToken)
     {
         ObjectDisposedException.ThrowIf(_disposed, this);
 
-        // Single-caller by contract: the reconnect supervisor (Task 4) owns connect/disconnect
+        // The options are rebuilt per attempt so a rotated CA file is picked up on the next
-        // sequencing, so no lock is taken here. The options are rebuilt per attempt so a rotated
+        // connect rather than being pinned for the life of the process.
-        // 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);
 
@@ -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();

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs

@@ -89,6 +89,45 @@ public sealed class MqttConnectionTests
         await Should.ThrowAsync<TimeoutException>(async () => await conn.ConnectAsync(CancellationToken.None));
     }
 
+    /// <summary>
+    ///     Task 4's reconnect loop retries connect on the SAME instance, reusing the
+    ///     <c>IMqttClient</c> held across failures. A failed attempt must therefore leave the
+    ///     connection usable rather than poisoned.
+    /// </summary>
+    [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<TimeoutException>(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<TimeoutException>(async () => await conn.ConnectAsync(CancellationToken.None));
+        conn.IsConnected.ShouldBeFalse();
+    }
+
+    /// <summary>
+    ///     Connecting a disposed instance is an <see cref="ObjectDisposedException"/> — the same
+    ///     classification the in-flight dispose race folds into. Dispose is idempotent.
+    /// </summary>
+    [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<ObjectDisposedException>(async () => await conn.ConnectAsync(CancellationToken.None));
+    }
+
     /// <summary>Never log or throw credentials — the plan's cross-cutting secrets rule.</summary>
     [Fact]
     public async Task ConnectAsync_Failure_DoesNotLeakPasswordIntoExceptionMessage()
@@ -153,21 +192,76 @@ public sealed class MqttConnectionTests
         var tls = built.ChannelOptions.ShouldBeOfType<MqttClientTcpOptions>().TlsOptions;
 
         tls.CertificateValidationHandler.ShouldNotBeNull();
-        tls.CertificateValidationHandler(SelfSignedValidationArgs(built)).ShouldBeFalse();
+        using var stranger = SelfSignedLeaf(Host);
+        tls.CertificateValidationHandler(ValidationArgs(built, stranger)).ShouldBeFalse();
     }
 
     /// <summary>An untrusted, unpinned server certificate is rejected — the pin actually validates.</summary>
     [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<MqttClientTcpOptions>().TlsOptions;
 
-            tls.CertificateValidationHandler!(SelfSignedValidationArgs(built)).ShouldBeFalse();
+            using var stranger = SelfSignedLeaf(Host);
+            tls.CertificateValidationHandler!(ValidationArgs(built, stranger)).ShouldBeFalse();
+        }
+        finally
+        {
+            File.Delete(caPath);
+        }
+    }
+
+    /// <summary>
+    ///     The accept path. Without this every certificate test asserts rejection, so deleting the
+    ///     <c>return true</c> — or getting <c>TrustMode</c> / <c>CustomTrustStore</c> wrong — would
+    ///     leave the suite green while no broker could ever connect.
+    /// </summary>
+    [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<MqttClientTcpOptions>().TlsOptions;
+
+            tls.CertificateValidationHandler!(ValidationArgs(built, leaf)).ShouldBeTrue();
+        }
+        finally
+        {
+            File.Delete(caPath);
+        }
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<MqttClientTcpOptions>().TlsOptions;
+
+            tls.CertificateValidationHandler!(ValidationArgs(built, leaf, intermediate)).ShouldBeTrue();
         }
         finally
         {
@@ -290,29 +384,87 @@ public sealed class MqttConnectionTests
         => MqttConnection.BuildClientOptions(opts, clientIdSuffix: null)
             .ChannelOptions.ShouldBeOfType<MqttClientTcpOptions>().TlsOptions;
 
-    private static MqttClientCertificateValidationEventArgs SelfSignedValidationArgs(MqttClientOptions built)
+    private static readonly DateTimeOffset NotBefore = DateTimeOffset.UtcNow.AddDays(-1);
+
+    /// <summary>A CA certificate, self-signed when <paramref name="issuer"/> is null.</summary>
+    private static X509Certificate2 IssueCa(X509Certificate2? issuer, string commonName, int validDays)
     {
-        using var rsa = RSA.Create(2048);
+        using var key = RSA.Create(2048);
-        var request = new CertificateRequest($"CN={Host}", rsa, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
+        var request = new CertificateRequest($"CN={commonName}", key, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
-        var cert = request.CreateSelfSigned(DateTimeOffset.UtcNow.AddDays(-1), DateTimeOffset.UtcNow.AddDays(1));
+        request.CertificateExtensions.Add(new X509BasicConstraintsExtension(true, false, 0, true));
-        return new MqttClientCertificateValidationEventArgs(
+        request.CertificateExtensions.Add(
-            cert,
+            new X509KeyUsageExtension(X509KeyUsageFlags.KeyCertSign | X509KeyUsageFlags.CrlSign, true));
-            new X509Chain(),
+        request.CertificateExtensions.Add(new X509SubjectKeyIdentifierExtension(request.PublicKey, false));
-            SslPolicyErrors.RemoteCertificateChainErrors,
+
-            built.ChannelOptions);
+        if (issuer is null)
+        {
+            return request.CreateSelfSigned(NotBefore, NotBefore.AddDays(validDays));
         }
 
-    private static string WriteTempPemCa()
+        // 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);
+    }
+
+    /// <summary>A server (end-entity) certificate signed by <paramref name="issuer"/>.</summary>
+    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());
+    }
+
+    /// <summary>A self-signed leaf that chains to nothing — the "stranger" a pin must reject.</summary>
+    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;
     }
 
+    /// <summary>
+    ///     Mimics what <c>SslStream</c> 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 <c>ChainElements</c> is populated the way the platform would populate it.
+    /// </summary>
+    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);
+    }
+
     /// <summary>
     ///     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