From 5629f3698d47247edae912b94b2ca8affe2b4f5e Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 24 Jul 2026 14:38:16 -0400 Subject: [PATCH] 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 --- .../MqttConnection.cs | 72 ++++++- .../MqttConnectionTests.cs | 188 ++++++++++++++++-- 2 files changed, 239 insertions(+), 21 deletions(-) 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