diff --git a/ZB.MOM.WW.OtOpcUa.slnx b/ZB.MOM.WW.OtOpcUa.slnx
index 6e6ccab6..e07dc176 100644
--- a/ZB.MOM.WW.OtOpcUa.slnx
+++ b/ZB.MOM.WW.OtOpcUa.slnx
@@ -42,6 +42,7 @@
+
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs
new file mode 100644
index 00000000..54337abe
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs
@@ -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;
+
+///
+/// Owns the driver's single live MQTTnet-5 client: assembles the broker connection options
+/// from (endpoint, protocol version, session, credentials,
+/// TLS + CA pin) and connects under a bounded deadline.
+///
+///
+///
+/// Connection-free construction. The constructor stores configuration only — it
+/// opens no socket and creates no client. 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.
+///
+///
+/// Credentials never leak. Nothing here logs or formats
+/// ; the options record itself redacts it in
+/// ToString().
+///
+/// Reconnect/backoff (Task 4) and subscription (Task 6) are deliberately not implemented here.
+///
+public sealed class MqttConnection : IAsyncDisposable
+{
+ private readonly string _driverId;
+ private readonly ILogger? _logger;
+ private readonly MqttDriverOptions _options;
+
+ private IMqttClient? _client;
+ private bool _disposed;
+
+ /// Stores configuration only — no network, no client instantiation.
+ /// 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;
+ }
+
+ /// Whether the underlying client currently holds an established MQTT session.
+ public bool IsConnected => _client?.IsConnected ?? false;
+
+ ///
+ 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();
+ }
+ }
+
+ ///
+ /// Connects to the broker, failing at
+ /// rather than waiting indefinitely on an unresponsive peer.
+ ///
+ /// Caller cancellation; linked with the connect deadline.
+ /// The connect deadline elapsed.
+ /// was cancelled.
+ 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);
+ }
+ }
+
+ ///
+ /// 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);
+ 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."),
+ };
+}
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj
new file mode 100644
index 00000000..55ffef0b
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj
@@ -0,0 +1,29 @@
+
+
+
+ net10.0
+ enable
+ enable
+ latest
+ true
+ true
+ $(NoWarn);CS1591
+ ZB.MOM.WW.OtOpcUa.Driver.Mqtt
+ ZB.MOM.WW.OtOpcUa.Driver.Mqtt
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
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
new file mode 100644
index 00000000..45180483
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs
@@ -0,0 +1,373 @@
+using System.Diagnostics;
+using System.Net;
+using System.Net.Security;
+using System.Net.Sockets;
+using System.Security.Cryptography;
+using System.Security.Cryptography.X509Certificates;
+using MQTTnet;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
+
+///
+/// Covers 's two Task-3 responsibilities: the pure
+/// options assembly (TLS / CA pin /
+/// credentials / protocol knobs) and the bounded
+/// deadline.
+///
+public sealed class MqttConnectionTests
+{
+ private const string Host = "broker.example.test";
+
+ // ---------------------------------------------------------------------------------
+ // Bounded connect (the R2-01 frozen-peer lesson: no unbounded wait on a dead broker)
+ // ---------------------------------------------------------------------------------
+
+ ///
+ /// A broker that completes the TCP handshake but never answers CONNACK is the shape that
+ /// actually wedges a client — a refused port returns ECONNREFUSED instantly and would pass
+ /// whether or not any deadline logic existed.
+ ///
+ [Fact]
+ public async Task ConnectAsync_BlackholeBroker_FailsWithinDeadline_NoHang()
+ {
+ using var blackhole = new BlackholeBroker();
+ var opts = new MqttDriverOptions
+ {
+ Host = "127.0.0.1", Port = blackhole.Port, UseTls = false, ConnectTimeoutSeconds = 2,
+ };
+ await using var conn = new MqttConnection(opts, driverId: "t", logger: null);
+
+ var sw = Stopwatch.StartNew();
+ await Should.ThrowAsync(async () => await conn.ConnectAsync(CancellationToken.None));
+ sw.Stop();
+
+ sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10));
+ conn.IsConnected.ShouldBeFalse();
+ }
+
+ ///
+ /// Falsifiability control for the linked-CTS wiring specifically: the configured connect
+ /// deadline is 60 s, so the only thing that can end this call inside 10 s is the caller's
+ /// token actually reaching MQTTnet. Drop the CreateLinkedTokenSource and this test
+ /// hangs for a minute.
+ ///
+ [Fact]
+ public async Task ConnectAsync_CallerCancellation_ObservedLongBeforeConnectDeadline()
+ {
+ using var blackhole = new BlackholeBroker();
+ var opts = new MqttDriverOptions
+ {
+ Host = "127.0.0.1", Port = blackhole.Port, UseTls = false, ConnectTimeoutSeconds = 60,
+ };
+ await using var conn = new MqttConnection(opts, driverId: "t", logger: null);
+ using var caller = new CancellationTokenSource(TimeSpan.FromMilliseconds(300));
+
+ var sw = Stopwatch.StartNew();
+ await Should.ThrowAsync(async () => await conn.ConnectAsync(caller.Token));
+ sw.Stop();
+
+ sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10));
+ }
+
+ ///
+ /// The deadline leg is distinguishable from a caller cancellation — a timed-out connect
+ /// surfaces , not a bare ,
+ /// so an operator can tell "broker never answered" from "we were asked to stop".
+ ///
+ [Fact]
+ public async Task ConnectAsync_DeadlineElapsed_ThrowsTimeoutException()
+ {
+ 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));
+ }
+
+ /// Never log or throw credentials — the plan's cross-cutting secrets rule.
+ [Fact]
+ public async Task ConnectAsync_Failure_DoesNotLeakPasswordIntoExceptionMessage()
+ {
+ const string secret = "sup3r-s3cret-broker-pw";
+ using var blackhole = new BlackholeBroker();
+ var opts = new MqttDriverOptions
+ {
+ Host = "127.0.0.1",
+ Port = blackhole.Port,
+ UseTls = false,
+ ConnectTimeoutSeconds = 1,
+ Username = "svc",
+ Password = secret,
+ };
+ await using var conn = new MqttConnection(opts, driverId: "t", logger: null);
+
+ var ex = await Should.ThrowAsync(async () => await conn.ConnectAsync(CancellationToken.None));
+
+ ex.ToString().ShouldNotContain(secret);
+ }
+
+ // ---------------------------------------------------------------------------------
+ // BuildClientOptions — pure assembly, no I/O
+ // ---------------------------------------------------------------------------------
+
+ [Fact]
+ public void BuildClientOptions_UseTlsWithCaPin_SetsTlsAndInstallsChainValidator()
+ {
+ var opts = new MqttDriverOptions
+ {
+ Host = Host,
+ Port = 8883,
+ UseTls = true,
+ AllowUntrustedServerCertificate = false,
+ CaCertificatePath = "/tmp/ca.pem",
+ };
+
+ var tls = BuildTls(opts);
+
+ tls.UseTls.ShouldBeTrue();
+ tls.TargetHost.ShouldBe(Host);
+ tls.AllowUntrustedCertificates.ShouldBeFalse();
+ tls.IgnoreCertificateChainErrors.ShouldBeFalse();
+ tls.CertificateValidationHandler.ShouldNotBeNull();
+ }
+
+ ///
+ /// The CA file is read lazily inside the validation callback, never by the builder — so a
+ /// path that does not exist yet must not throw at build time, and must fail the *connection*
+ /// closed when the handshake finally asks.
+ ///
+ [Fact]
+ public void BuildClientOptions_CaPinPathMissing_BuildsWithoutIo_AndValidatorFailsClosed()
+ {
+ var missing = Path.Combine(Path.GetTempPath(), $"otopcua-mqtt-no-such-ca-{Guid.NewGuid():N}.pem");
+ File.Exists(missing).ShouldBeFalse();
+
+ var opts = new MqttDriverOptions { Host = Host, Port = 8883, UseTls = true, CaCertificatePath = missing };
+
+ var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
+ var tls = built.ChannelOptions.ShouldBeOfType().TlsOptions;
+
+ tls.CertificateValidationHandler.ShouldNotBeNull();
+ tls.CertificateValidationHandler(SelfSignedValidationArgs(built)).ShouldBeFalse();
+ }
+
+ /// An untrusted, unpinned server certificate is rejected — the pin actually validates.
+ [Fact]
+ public void BuildClientOptions_CaPin_RejectsCertificateNotIssuedByThePinnedCa()
+ {
+ var caPath = WriteTempPemCa();
+ 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();
+ }
+ finally
+ {
+ File.Delete(caPath);
+ }
+ }
+
+ /// Default posture: OS trust store, no custom handler, escape hatch off.
+ [Fact]
+ public void BuildClientOptions_TlsWithoutCaPin_LeavesOsTrustStoreValidationInPlace()
+ {
+ var opts = new MqttDriverOptions { Host = Host, Port = 8883, UseTls = true };
+
+ var tls = BuildTls(opts);
+
+ tls.UseTls.ShouldBeTrue();
+ tls.CertificateValidationHandler.ShouldBeNull();
+ tls.AllowUntrustedCertificates.ShouldBeFalse();
+ tls.IgnoreCertificateChainErrors.ShouldBeFalse();
+ tls.IgnoreCertificateRevocationErrors.ShouldBeFalse();
+ }
+
+ /// The dev/on-prem escape hatch only opens when explicitly asked for.
+ [Fact]
+ public void BuildClientOptions_AllowUntrustedServerCertificate_OpensEscapeHatch()
+ {
+ var opts = new MqttDriverOptions
+ {
+ Host = Host, Port = 8883, UseTls = true, AllowUntrustedServerCertificate = true,
+ };
+
+ var tls = BuildTls(opts);
+
+ tls.AllowUntrustedCertificates.ShouldBeTrue();
+ tls.IgnoreCertificateChainErrors.ShouldBeTrue();
+ tls.CertificateValidationHandler.ShouldNotBeNull();
+ }
+
+ [Fact]
+ public void BuildClientOptions_UseTlsFalse_KeepsTlsOffAndInstallsNoValidator()
+ {
+ var opts = new MqttDriverOptions
+ {
+ Host = Host, Port = 1883, UseTls = false, AllowUntrustedServerCertificate = true, CaCertificatePath = "/tmp/ca.pem",
+ };
+
+ var tls = BuildTls(opts);
+
+ tls.UseTls.ShouldBeFalse();
+ tls.CertificateValidationHandler.ShouldBeNull();
+ tls.AllowUntrustedCertificates.ShouldBeFalse();
+ }
+
+ [Fact]
+ public void BuildClientOptions_MapsProtocolAndSessionKnobs()
+ {
+ var opts = new MqttDriverOptions
+ {
+ Host = Host,
+ Port = 8883,
+ ClientId = "otopcua-node-1",
+ ProtocolVersion = MqttProtocolVersion.V311,
+ CleanSession = false,
+ KeepAliveSeconds = 45,
+ ConnectTimeoutSeconds = 7,
+ };
+
+ var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
+
+ built.ProtocolVersion.ShouldBe(MQTTnet.Formatter.MqttProtocolVersion.V311);
+ built.ClientId.ShouldBe("otopcua-node-1");
+ built.CleanSession.ShouldBeFalse();
+ built.KeepAlivePeriod.ShouldBe(TimeSpan.FromSeconds(45));
+ built.Timeout.ShouldBe(TimeSpan.FromSeconds(7));
+ var tcp = built.ChannelOptions.ShouldBeOfType();
+ tcp.RemoteEndpoint.ShouldBeOfType().Host.ShouldBe(Host);
+ tcp.RemoteEndpoint.ShouldBeOfType().Port.ShouldBe(8883);
+ }
+
+ [Fact]
+ public void BuildClientOptions_ClientIdSuffix_IsAppended()
+ {
+ var opts = new MqttDriverOptions { Host = Host, ClientId = "otopcua" };
+
+ MqttConnection.BuildClientOptions(opts, clientIdSuffix: "-browse").ClientId.ShouldBe("otopcua-browse");
+ }
+
+ [Fact]
+ public void BuildClientOptions_NoUsername_SendsNoCredentials()
+ => MqttConnection.BuildClientOptions(new MqttDriverOptions { Host = Host }, clientIdSuffix: null)
+ .Credentials.ShouldBeNull();
+
+ [Fact]
+ public void BuildClientOptions_Username_SendsCredentials()
+ {
+ var opts = new MqttDriverOptions { Host = Host, Username = "svc", Password = "pw" };
+
+ var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
+
+ built.Credentials.ShouldNotBeNull();
+ built.Credentials.GetUserName(built).ShouldBe("svc");
+ }
+
+ /// The ctor must not touch the network — the universal-browser throwaway-instance pattern.
+ [Fact]
+ public void Ctor_IsConnectionFree()
+ {
+ var opts = new MqttDriverOptions { Host = "127.0.0.1", Port = 1, UseTls = false };
+
+ var conn = new MqttConnection(opts, driverId: "t", logger: null);
+
+ conn.IsConnected.ShouldBeFalse();
+ }
+
+ // ---------------------------------------------------------------------------------
+ // helpers
+ // ---------------------------------------------------------------------------------
+
+ private static MqttClientTlsOptions BuildTls(MqttDriverOptions opts)
+ => MqttConnection.BuildClientOptions(opts, clientIdSuffix: null)
+ .ChannelOptions.ShouldBeOfType().TlsOptions;
+
+ private static MqttClientCertificateValidationEventArgs SelfSignedValidationArgs(MqttClientOptions built)
+ {
+ 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);
+ }
+
+ private static string WriteTempPemCa()
+ {
+ 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());
+ return path;
+ }
+
+ ///
+ /// 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
+ /// not (it refuses instantly and proves nothing about a deadline).
+ ///
+ private sealed class BlackholeBroker : IDisposable
+ {
+ private readonly List _accepted = [];
+ private readonly CancellationTokenSource _cts = new();
+ private readonly TcpListener _listener;
+
+ public BlackholeBroker()
+ {
+ _listener = new TcpListener(IPAddress.Loopback, 0);
+ _listener.Start();
+ Port = ((IPEndPoint)_listener.LocalEndpoint).Port;
+ _ = Task.Run(AcceptLoopAsync);
+ }
+
+ public int Port { get; }
+
+ public void Dispose()
+ {
+ _cts.Cancel();
+ _listener.Stop();
+ lock (_accepted)
+ {
+ foreach (var client in _accepted)
+ {
+ client.Dispose();
+ }
+
+ _accepted.Clear();
+ }
+
+ _cts.Dispose();
+ }
+
+ private async Task AcceptLoopAsync()
+ {
+ try
+ {
+ while (!_cts.IsCancellationRequested)
+ {
+ var client = await _listener.AcceptTcpClientAsync(_cts.Token);
+ lock (_accepted)
+ {
+ _accepted.Add(client);
+ }
+ }
+ }
+ catch (Exception)
+ {
+ // Listener stopped / token cancelled — the fixture is going away.
+ }
+ }
+ }
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj
index 65591a1b..2ad93142 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj
@@ -21,6 +21,7 @@
+