dohertj2/lmxopcua
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request '[opcuaclient] OpcUaClient — Reverse Connect' (#392) from auto/opcuaclient/11 into auto/driver-gaps

Browse Source
This commit was merged in pull request #392.
This commit is contained in:
dohertj2
2026-04-26 06:11:11 -04:00
parent 9a3bc08e1c 5c72deb839
commit ac3fd45cc6
10 changed files with 920 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
docs/drivers/OpcUaClient-Test-Fixture.md
View File

@@ -47,6 +47,13 @@ the tests mock.
- `OpcUaClientSmokeTests.Client_subscribe_receives_StepUp_data_changes_from_live_server` —
real `MonitoredItem` subscription against `ns=3;s=FastUInt1` (ticks every
100 ms); asserts `OnDataChange` fires within 3 s of subscribe
- `OpcUaClientReverseConnectSmokeTests.Driver_accepts_reverse_connect_from_opc_plc_rc_simulator` —
reverse-connect (server-initiated) coverage. Driver binds
`opc.tcp://0.0.0.0:4844`, the `opc-plc-rc` docker service dials in via
`--rc opc.tcp://host.docker.internal:4844`, and a Read round-trips over
the inbound socket. Gated on `OPCUA_RC_SIM=1` because the simulator
requires `host.docker.internal` resolution which not every CI runner
exposes.
Wire-level surfaces verified: `IDriver` + `IReadable` + `ISubscribable` +
`IHostConnectivityProbe` (via the Secure Channel exchange).

67
docs/drivers/OpcUaClient.md
View File

@@ -59,3 +59,70 @@ Flip `WatchModelChanges` to `false` when:
- The upstream server fires spurious `ModelChangeEvent`s that don't
reflect real topology changes, causing wasted re-imports. Tighten or
disable rather than chasing the noise downstream.
## Reverse Connect (server-initiated)
OPC UA's reverse-connect mode flips the transport direction: instead of the
client dialling the server, the **server** dials the client's listener. The
upstream sends a `ReverseHello` and the client continues the OPC UA
handshake on the inbound socket. Required for OT-DMZ deployments where the
plant firewall only permits outbound traffic from the upstream — the
gateway opens a listener, the upstream reaches out.
### Configuration
| Option | Default | Notes |
| --- | --- | --- |
| `ReverseConnect.Enabled` | `false` | Opt-in. When `true`, replaces the failover dial-sweep with a `WaitForConnection` call. |
| `ReverseConnect.ListenerUrl` | `null` | Local listener URL the SDK binds. Typically `opc.tcp://0.0.0.0:4844` (any interface) or a specific NIC for multi-homed gateways. **Required when `Enabled` is `true`.** |
| `ReverseConnect.ExpectedServerUri` | `null` | Upstream's `ApplicationUri` to filter inbound dials. `null` accepts the first connection (only safe with one upstream targeting the listener). |
### Shared listener (singleton)
A single underlying `Opc.Ua.Client.ReverseConnectManager` per process keyed
on `ListenerUrl`. Two driver instances that share a listener URL multiplex
onto one TCP socket; the SDK demuxes inbound dials by the upstream's
reported `ServerUri`. The wrapper (`ReverseConnectListener`) is
reference-counted — first `Acquire` binds the port, last `Release` tears it
down. Letting drivers come and go independently without races on
port-bind / port-unbind.
When two drivers share a listener:
- They MUST set `ExpectedServerUri` to disambiguate; otherwise the first
upstream to dial in wins regardless of which driver is waiting.
- They CAN come and go independently; the listener stays alive while at
least one driver references it.
### Behaviour
- The dial path is bypassed entirely when `Enabled` is `true`. Failover
across multiple `EndpointUrls` doesn't apply — there's no client-side
dial to fail over.
- `ExpectedServerUri` is the SDK's filter parameter to `WaitForConnectionAsync`.
Inbound `ReverseHello`s from a different upstream are ignored and the
caller keeps waiting.
- The same `EndpointDescription` derivation runs as the dial path — the
first `EndpointUrl` in the candidate list seeds `SecurityPolicy` /
`SecurityMode` / `EndpointUrl` for the session-create call. The actual
endpoint lives on the upstream and the SDK reconciles after the
`ReverseHello`.
- Cancellation: `Timeout` bounds the wait. A stuck listener with no inbound
dial throws after `Timeout` rather than hanging init forever.
- Shutdown releases the listener reference. The last release stops the
listener so the port can be re-bound by a future driver lifecycle.
### Wiring it up on the upstream
The upstream OPC UA server has to be configured to dial out. The `opc-plc`
simulator does this with `--rc=opc.tcp://<gateway-host>:4844`; for a real
upstream see your server's reverse-connect docs (most major implementations
expose a "ReverseConnect.Endpoint" config knob).
### When NOT to use
- Standard plant networks where the gateway can dial the upstream — the
conventional dial path is simpler and supports failover natively.
- Public-internet OPC UA: reverse-connect is a network-policy workaround,
not a security primitive. Always pair with `Sign` or `SignAndEncrypt`
+ a vetted user-token policy.

30
scripts/e2e/test-opcuaclient.ps1
View File

@@ -54,8 +54,29 @@
.PARAMETER ChangeWaitSec
How long the subscribe stage waits for a data-change. Default 10s.
.PARAMETER ReverseConnect
When set, the script asserts the gateway is configured for reverse-connect
(server-initiated) mode. The OtOpcUa server's DriverConfig for the OpcUaClient
instance must already have ReverseConnect.Enabled=true + ListenerUrl set; this
script doesn't reconfigure the driver, only verifies the bridged path still
reads end-to-end with the listener up. The reverse-connect topology is opaque
to the downstream OPC UA client (us), so the read assertion is identical to
the dial-mode path — the value of running the script in this mode is to catch
regressions where reverse-connect breaks the post-init capability surface.
.PARAMETER ReverseListenerUrl
Documentation-only. The listener URL the gateway is expected to be bound to
when -ReverseConnect is set; printed in the run banner so operators can
cross-check their server config. Default opc.tcp://0.0.0.0:4844.
.EXAMPLE
.\test-opcuaclient.ps1 -BridgedNodeId "ns=2;s=/warsaw/opc-plc/StepUp"
.EXAMPLE
# OT-DMZ deployment: the upstream dials the gateway. The script flow is the
# same — we still drive the bridged read through the OtOpcUa server — but the
# banner reflects the reverse-connect topology.
.\test-opcuaclient.ps1 -BridgedNodeId "ns=2;s=/warsaw/opc-plc/StepUp" -ReverseConnect
#>
param(
@@ -63,7 +84,9 @@ param(
[string]$UpstreamUrl = "opc.tcp://localhost:50000",
[Parameter(Mandatory)] [string]$BridgedNodeId,
[string]$UpstreamNodeId = "ns=3;s=StepUp",
[int]$ChangeWaitSec = 10
[int]$ChangeWaitSec = 10,
[switch]$ReverseConnect,
[string]$ReverseListenerUrl = "opc.tcp://0.0.0.0:4844"
)
$ErrorActionPreference = "Stop"
@@ -73,6 +96,11 @@ $opcUaCli = Get-CliInvocation `
-ProjectFolder "src/ZB.MOM.WW.OtOpcUa.Client.CLI" `
-ExeName "otopcua-cli"
if ($ReverseConnect) {
Write-Host "[INFO] -ReverseConnect set: gateway is expected to be bound to listener $ReverseListenerUrl"
Write-Host "[INFO] Upstream OPC UA server should be configured with --rc=$ReverseListenerUrl (or equivalent on a real server)"
}
$results = @()
# Stage 1: probe

210
src/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs
View File

@@ -131,6 +131,34 @@ public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string d
private bool _disposed;
/// <summary>URL of the endpoint the driver actually connected to. Exposed via <see cref="HostName"/>.</summary>
private string? _connectedEndpointUrl;
/// <summary>
/// Reverse-connect listener acquired during <see cref="InitializeAsync"/> when
/// <see cref="ReverseConnectOptions.Enabled"/> is set. Null when reverse-connect is
/// disabled. Released back to the singleton pool on shutdown so multiple driver
/// instances on the same listener URL can come and go independently.
/// </summary>
private ReverseConnectListener? _reverseListener;
/// <summary>
/// Test seam — pluggable reverse-connect "wait" hook. When non-null,
/// <see cref="OpenReverseConnectSessionAsync"/> uses this delegate instead of
/// calling into a real <see cref="ReverseConnectListener"/>. Lets unit tests
/// inject a synthetic <c>ITransportWaitingConnection</c> without binding a port
/// or running the SDK's listener threads.
/// </summary>
internal Func<Uri, string?, CancellationToken, Task<Opc.Ua.ITransportWaitingConnection>>? ReverseConnectWaitHookForTest { get; set; }
/// <summary>
/// Test seam — pluggable session factory invoked in the reverse-connect path.
/// Tests can use this to verify that the session-create call receives the
/// expected <c>ITransportWaitingConnection</c> without instantiating the SDK
/// <see cref="DefaultSessionFactory"/> (which hits real cert + transport code).
/// </summary>
internal Func<ApplicationConfiguration, Opc.Ua.ITransportWaitingConnection, ConfiguredEndpoint, UserIdentity, CancellationToken, Task<ISession>>? ReverseConnectSessionFactoryForTest { get; set; }
/// <summary>Test seam — last reverse-connect listener acquired (null when reverse-connect is disabled or shut down).</summary>
internal ReverseConnectListener? ReverseListenerForTest => _reverseListener;
/// <summary>
/// SDK-provided reconnect handler that owns the retry loop + session-transfer machinery
/// when the session's keep-alive channel reports a bad status. Null outside the
@@ -202,34 +230,60 @@ public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string d
var identity = BuildUserIdentity(_options);
// Failover sweep: try each endpoint in order, return the session from the first
// one that successfully connects. Per-endpoint failures are captured so the final
// aggregate exception names every URL that was tried and why — critical diag for
// operators debugging 'why did the failover pick #3?'.
var attemptErrors = new List<string>(candidates.Count);
ISession? session = null;
string? connectedUrl = null;
foreach (var url in candidates)
{
try
{
session = await OpenSessionOnEndpointAsync(
appConfig, url, _options.SecurityPolicy, _options.SecurityMode,
identity, cancellationToken).ConfigureAwait(false);
connectedUrl = url;
break;
}
catch (Exception ex)
{
attemptErrors.Add($"{url} -> {ex.GetType().Name}: {ex.Message}");
}
}
if (session is null)
throw new AggregateException(
"OPC UA Client failed to connect to any of the configured endpoints. " +
"Tried:\n " + string.Join("\n ", attemptErrors),
attemptErrors.Select(e => new InvalidOperationException(e)));
if (_options.ReverseConnect.Enabled)
{
// Reverse-connect path: instead of dialling each candidate URL, we register
// our listener URL with the process-wide ReverseConnectManager and wait for
// the upstream server to dial in. The first candidate URL still drives
// EndpointDescription selection so SecurityPolicy/Mode + user-identity flow
// through the same code path as the conventional dial — only the transport
// direction flips. ExpectedServerUri filters incoming connections so the
// listener can be shared across drivers targeting different upstreams.
if (string.IsNullOrWhiteSpace(_options.ReverseConnect.ListenerUrl))
throw new InvalidOperationException(
"ReverseConnect.Enabled=true but ReverseConnect.ListenerUrl is not set. " +
"Configure a listener URL like 'opc.tcp://0.0.0.0:4844' so the upstream server can dial in.");
var endpointForReverse = candidates.FirstOrDefault()
?? throw new InvalidOperationException(
"ReverseConnect requires at least one EndpointUrl in the candidate list to derive the EndpointDescription from.");
session = await OpenReverseConnectSessionAsync(
appConfig, endpointForReverse, identity, cancellationToken).ConfigureAwait(false);
connectedUrl = endpointForReverse;
}
else
{
// Failover sweep: try each endpoint in order, return the session from the first
// one that successfully connects. Per-endpoint failures are captured so the final
// aggregate exception names every URL that was tried and why — critical diag for
// operators debugging 'why did the failover pick #3?'.
var attemptErrors = new List<string>(candidates.Count);
foreach (var url in candidates)
{
try
{
session = await OpenSessionOnEndpointAsync(
appConfig, url, _options.SecurityPolicy, _options.SecurityMode,
identity, cancellationToken).ConfigureAwait(false);
connectedUrl = url;
break;
}
catch (Exception ex)
{
attemptErrors.Add($"{url} -> {ex.GetType().Name}: {ex.Message}");
}
}
if (session is null)
throw new AggregateException(
"OPC UA Client failed to connect to any of the configured endpoints. " +
"Tried:\n " + string.Join("\n ", attemptErrors),
attemptErrors.Select(e => new InvalidOperationException(e)));
}
// Wire the session's keep-alive channel into HostState + the reconnect trigger.
// OPC UA keep-alives are authoritative for session liveness: the SDK pings on
@@ -268,6 +322,13 @@ public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string d
{
try { if (Session is Session s) await s.CloseAsync().ConfigureAwait(false); } catch { }
Session = null;
// Release the reverse-connect listener if we acquired it but session-create failed
// — leaks a port-bind otherwise, blocking the next init attempt.
if (_reverseListener is not null)
{
try { _reverseListener.Release(); } catch { /* best-effort */ }
_reverseListener = null;
}
_health = new DriverHealth(DriverState.Faulted, null, ex.Message);
throw;
}
@@ -644,6 +705,96 @@ public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string d
return session;
}
/// <summary>
/// Open a session over a server-initiated (reverse) connect. Acquires a process-wide
/// <see cref="ReverseConnectListener"/> for the configured listener URL, waits for the
/// upstream server to dial in (filtered by <see cref="ReverseConnectOptions.ExpectedServerUri"/>),
/// then hands the resulting <see cref="Opc.Ua.ITransportWaitingConnection"/> into the
/// session-create path. The endpoint description still comes from the candidate URL so
/// SecurityPolicy / Mode / cert handling are identical to the dial path — only the
/// transport direction flips.
/// </summary>
private async Task<ISession> OpenReverseConnectSessionAsync(
ApplicationConfiguration appConfig,
string endpointUrl,
UserIdentity identity,
CancellationToken ct)
{
var listenerUrl = _options.ReverseConnect.ListenerUrl!;
var expectedServerUri = _options.ReverseConnect.ExpectedServerUri;
// Acquire a ref to the singleton listener for this URL. Multiple driver instances
// sharing a URL share one underlying SDK manager — see ReverseConnectListener for
// the ref-count model.
if (ReverseConnectWaitHookForTest is null)
{
_reverseListener = ReverseConnectListener.Acquire(listenerUrl, appConfig);
}
// Build the ConfiguredEndpoint from the configured endpointUrl. We DON'T call
// GetEndpointsAsync over reverse connect here — the SDK's reverse-connect overload
// accepts a synthetic EndpointDescription and the upstream resends its real one
// during ReverseHello, so a static description is fine for the create call.
var endpointDescription = new EndpointDescription(endpointUrl)
{
SecurityPolicyUri = MapSecurityPolicy(_options.SecurityPolicy),
SecurityMode = _options.SecurityMode switch
{
OpcUaSecurityMode.None => MessageSecurityMode.None,
OpcUaSecurityMode.Sign => MessageSecurityMode.Sign,
OpcUaSecurityMode.SignAndEncrypt => MessageSecurityMode.SignAndEncrypt,
_ => MessageSecurityMode.None,
},
};
var endpointConfig = EndpointConfiguration.Create(appConfig);
endpointConfig.OperationTimeout = (int)_options.Timeout.TotalMilliseconds;
var endpoint = new ConfiguredEndpoint(null, endpointDescription, endpointConfig);
// Wait for the upstream to dial in. Bounded by Timeout so a stuck listener doesn't
// hang init forever — operators see a clear timeout error rather than a silent stall.
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
cts.CancelAfter(_options.Timeout);
Opc.Ua.ITransportWaitingConnection connection;
if (ReverseConnectWaitHookForTest is not null)
{
connection = await ReverseConnectWaitHookForTest(
new Uri(listenerUrl), expectedServerUri, cts.Token).ConfigureAwait(false);
}
else
{
connection = await _reverseListener!.WaitForServerAsync(
new Uri(listenerUrl), expectedServerUri, cts.Token).ConfigureAwait(false);
}
// Hand the inbound connection into the session-create path. The factory hook lets
// unit tests assert that the right connection + endpoint flow through without
// standing up a real DefaultSessionFactory (which expects a fully-wired transport).
ISession session;
if (ReverseConnectSessionFactoryForTest is not null)
{
session = await ReverseConnectSessionFactoryForTest(
appConfig, connection, endpoint, identity, cts.Token).ConfigureAwait(false);
}
else
{
session = await new DefaultSessionFactory(telemetry: null!).CreateAsync(
appConfig,
connection,
endpoint,
updateBeforeConnect: false,
checkDomain: false,
_options.SessionName,
(uint)_options.SessionTimeout.TotalMilliseconds,
identity,
preferredLocales: null,
cts.Token).ConfigureAwait(false);
}
session.KeepAliveInterval = (int)_options.KeepAliveInterval.TotalMilliseconds;
return session;
}
/// <summary>
/// Select the remote endpoint matching both the requested <paramref name="policy"/>
/// and <paramref name="mode"/>. The SDK's <c>CoreClientUtils.SelectEndpointAsync</c>
@@ -799,6 +950,15 @@ public sealed class OpcUaClientDriver(OpcUaClientDriverOptions options, string d
_connectedEndpointUrl = null;
_operationLimits = null;
// Release our hold on the reverse-connect listener. Last release tears the manager
// down; siblings that share the URL keep it alive. Idempotent — releasing a null
// listener (e.g. shutdown after a failed init) is a no-op.
if (_reverseListener is not null)
{
try { _reverseListener.Release(); } catch { /* best-effort */ }
_reverseListener = null;
}
TransitionTo(HostState.Unknown);
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
}

58
src/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriverOptions.cs
View File

@@ -253,8 +253,66 @@ public sealed class OpcUaClientDriverOptions
/// short enough that single-node adds re-import promptly.
/// </summary>
public TimeSpan ModelChangeDebounce { get; init; } = TimeSpan.FromSeconds(5);
/// <summary>
/// Reverse-connect (server-initiated) configuration. When
/// <see cref="ReverseConnectOptions.Enabled"/> is <c>true</c> the driver flips the
/// transport direction: instead of dialling the upstream server, it opens a TCP
/// listener on <see cref="ReverseConnectOptions.ListenerUrl"/> and waits for the
/// upstream server to initiate the connection ("ReverseHello"). Required for
/// OT-DMZ deployments where the firewall only permits outbound traffic from the
/// plant network — the upstream server reaches out, the gateway listens.
/// </summary>
public ReverseConnectOptions ReverseConnect { get; init; } = new();
}
/// <summary>
/// Driver knobs for OPC UA reverse-connect (server-initiated) sessions. Mirrors the
/// SDK's <c>Opc.Ua.Client.ReverseConnectManager</c> surface but expressed as plain
/// config so the driver can decide listener-mode vs dial-mode at startup.
/// </summary>
/// <remarks>
/// <para>
/// <b>Direction</b>: in conventional OPC UA the client opens the TCP connection
/// to the server. Reverse-connect inverts this — the server initiates a TCP
/// connection to a listener the client exposes, then sends a <c>ReverseHello</c>
/// message naming itself; the client picks the right session config and continues
/// the OPC UA handshake on the inbound socket. Critical for OT-DMZ networks where
/// the plant firewall only allows outbound traffic.
/// </para>
/// <para>
/// <b>Singleton listener</b>: a single <c>ReverseConnectManager</c> per process
/// keyed on <see cref="ListenerUrl"/> multiplexes inbound connections across
/// driver instances. Two drivers configured with the same listener URL share one
/// underlying TCP socket; the manager dispatches by the upstream's reported
/// <c>ServerUri</c>. See <c>ReverseConnectListener</c>.
/// </para>
/// </remarks>
/// <param name="Enabled">
/// When <c>true</c>, the driver opens a listener at <see cref="ListenerUrl"/> and
/// waits for the upstream server to initiate the session. When <c>false</c>
/// (default), the driver uses the conventional dial path against
/// <see cref="OpcUaClientDriverOptions.EndpointUrls"/>.
/// </param>
/// <param name="ListenerUrl">
/// Local listener URL the SDK binds when reverse-connect is enabled. Typically
/// <c>opc.tcp://0.0.0.0:4844</c> to accept on every interface, or pinned to a
/// specific NIC for multi-homed gateways. Required when
/// <see cref="Enabled"/> is <c>true</c>.
/// </param>
/// <param name="ExpectedServerUri">
/// The upstream server's <c>ApplicationUri</c> the driver expects to see in the
/// <c>ReverseHello</c>. The SDK passes this as the <c>serverUri</c> filter to
/// <c>WaitForConnectionAsync</c> so connections from a different upstream are ignored
/// — important when the listener is shared across multiple drivers and several
/// upstreams might dial in. Leave <c>null</c> to accept the first connection regardless
/// (only safe when exactly one upstream targets the listener).
/// </param>
public sealed record ReverseConnectOptions(
bool Enabled = false,
string? ListenerUrl = null,
string? ExpectedServerUri = null);
/// <summary>
/// Selective import + namespace remap rules for the OPC UA Client driver. Pure local
/// filtering inside <c>BrowseRecursiveAsync</c> + <c>EnrichAndRegisterVariablesAsync</c>;

156
src/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/ReverseConnectListener.cs Normal file
View File

@@ -0,0 +1,156 @@
using System.Collections.Concurrent;
using Opc.Ua;
using Opc.Ua.Client;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;
/// <summary>
/// Process-wide ref-counted wrapper around <see cref="ReverseConnectManager"/>.
/// Multiple <see cref="OpcUaClientDriver"/> instances that share a listener URL
/// multiplex onto a single underlying manager — opening N sockets on the same
/// port would conflict at the OS layer, and the SDK's manager already dispatches
/// inbound connections by <c>ServerUri</c>, so one manager per listener URL is
/// both correct and unavoidable.
/// </summary>
/// <remarks>
/// <para>
/// <b>Lifecycle</b>: callers <see cref="Acquire"/> a listener for their
/// configured URL during driver init and <see cref="Release"/> on shutdown.
/// The first <see cref="Acquire"/> for a given URL constructs the manager,
/// calls <c>AddEndpoint</c>, and starts the listener. The last
/// <see cref="Release"/> stops the listener + disposes the manager. Reference
/// counting lets two drivers come up + go down independently without racing on
/// port-bind / port-unbind.
/// </para>
/// <para>
/// <b>Why not Lazy&lt;T&gt;</b>: the listener key is dynamic (the URL string)
/// which Lazy doesn't model. A <see cref="ConcurrentDictionary{TKey,TValue}"/>
/// + locked entry counter is the simplest pattern that handles add/remove
/// atomically without a global lock.
/// </para>
/// </remarks>
internal sealed class ReverseConnectListener : IDisposable
{
private static readonly ConcurrentDictionary<string, ReverseConnectListener> s_instances
= new(StringComparer.OrdinalIgnoreCase);
private static readonly object s_globalLock = new();
private readonly string _listenerUrl;
private readonly ReverseConnectManager _manager;
private int _refCount;
private bool _started;
private ApplicationConfiguration? _appConfig;
/// <summary>
/// Test seam — total instances ever created for a given URL. Lets unit tests
/// assert that two drivers sharing a URL only spin up one underlying manager.
/// </summary>
internal static int InstanceCountForTest(string listenerUrl) =>
s_instances.TryGetValue(listenerUrl, out var inst) ? inst._refCount : 0;
/// <summary>Test seam — peek the underlying SDK manager for assertion / mocking.</summary>
internal ReverseConnectManager Manager => _manager;
/// <summary>Test seam — current refcount on this entry.</summary>
internal int RefCountForTest => _refCount;
private ReverseConnectListener(string listenerUrl)
{
_listenerUrl = listenerUrl;
#pragma warning disable CS0618 // ITelemetryContext-less ctor is obsolete; the driver doesn't plumb telemetry
_manager = new ReverseConnectManager();
#pragma warning restore CS0618
}
/// <summary>
/// Acquire a reference to the listener for <paramref name="listenerUrl"/>. The
/// first acquire constructs the manager + binds the socket; subsequent acquires
/// bump the ref-count and reuse the live manager. Idempotent on
/// <c>AddEndpoint</c> — the SDK's manager tolerates the same URL being added
/// twice but the listener guards against it anyway so the wire-side state stays
/// deterministic across test runs. <paramref name="appConfig"/> is needed by
/// <c>StartService</c> so the manager has the cert validator + transport quotas
/// of the calling driver. The first acquire's config wins — subsequent acquires
/// trust that all drivers sharing the listener use compatible cert / quota config.
/// </summary>
public static ReverseConnectListener Acquire(string listenerUrl, ApplicationConfiguration appConfig)
{
if (string.IsNullOrWhiteSpace(listenerUrl))
throw new ArgumentException("listenerUrl is required", nameof(listenerUrl));
ArgumentNullException.ThrowIfNull(appConfig);
lock (s_globalLock)
{
var entry = s_instances.GetOrAdd(listenerUrl, u => new ReverseConnectListener(u));
entry._refCount++;
entry.EnsureStarted(appConfig);
return entry;
}
}
private void EnsureStarted(ApplicationConfiguration appConfig)
{
if (_started) return;
_appConfig = appConfig;
var uri = new Uri(_listenerUrl);
_manager.AddEndpoint(uri);
_manager.StartService(appConfig);
_started = true;
}
/// <summary>
/// Test seam — construct a listener entry without binding a real socket. Used
/// by unit tests to verify acquire/release ref-counting without paying the
/// SDK + OS port-bind cost.
/// </summary>
internal static ReverseConnectListener AcquireForTest(string listenerUrl)
{
if (string.IsNullOrWhiteSpace(listenerUrl))
throw new ArgumentException("listenerUrl is required", nameof(listenerUrl));
lock (s_globalLock)
{
var entry = s_instances.GetOrAdd(listenerUrl, u => new ReverseConnectListener(u));
entry._refCount++;
// Skip EnsureStarted — tests don't want the real port bind.
return entry;
}
}
/// <summary>
/// Wait for the upstream named by <paramref name="serverUri"/> to dial in to
/// this listener. Proxies straight to the SDK's
/// <c>WaitForConnection</c> — the listener owns lifecycle, the SDK owns the
/// wait + the inbound transport object.
/// </summary>
public Task<ITransportWaitingConnection> WaitForServerAsync(
Uri endpointUrl,
string? serverUri,
CancellationToken ct = default) =>
_manager.WaitForConnectionAsync(endpointUrl, serverUri, ct);
/// <summary>
/// Drop one reference. When the last reference goes away the manager is stopped
/// and the dictionary entry removed so a future driver can rebind the same port
/// cleanly. Releasing a listener that's already been fully torn down is a no-op
/// so disposal paths can be defensive.
/// </summary>
public void Release()
{
lock (s_globalLock)
{
_refCount--;
if (_refCount > 0) return;
// Dispose() walks DisposeHosts() → CloseHosts() internally so the listener
// socket is released cleanly. There's no public StopService; the SDK exposes
// only Dispose() for external teardown.
try { _manager.Dispose(); } catch { /* best-effort during shutdown */ }
s_instances.TryRemove(_listenerUrl, out _);
_started = false;
}
}
public void Dispose() => Release();
}

38
tests/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests/Docker/docker-compose.yml
View File

@@ -43,3 +43,41 @@ services:
timeout: 2s
retries: 10
start_period: 10s
# opc-plc-rc — reverse-connect (server-initiated) variant. The simulator
# acts as the OPC UA server but, unlike the regular service above, it dials
# OUT to the client's listener URL instead of accepting an inbound dial.
# Mirrors the OT-DMZ topology where the plant firewall only permits
# outbound traffic from the upstream server. The driver-side test fixture
# binds opc.tcp://0.0.0.0:4844 and waits for opc-plc-rc to ReverseHello.
#
# `--rc` is opc-plc's reverse-connect knob — value is the client URL the
# simulator should dial when it has no inbound connection. host.docker.internal
# is the docker-for-windows / docker-for-mac shorthand for the host's IP;
# on Linux hosts use --add-host=host.docker.internal:host-gateway.
opc-plc-rc:
image: mcr.microsoft.com/iotedge/opc-plc:2.14.10
container_name: otopcua-opc-plc-rc
restart: "no"
extra_hosts:
- "host.docker.internal:host-gateway"
command:
# --pn=50001: bind on a different port so this container can run alongside
# the dial-mode simulator above. Reverse-connect doesn't require
# the client to know this port (the simulator is the dialer)
# but it still has to bind one for any incoming admin queries.
# --rc: reverse-connect target — the simulator dials this URL and
# presents its OPC UA endpoint over the inbound socket. Must
# point at the test runner's listener.
# --ut/--aa/--alm: same flags as the regular profile.
- "--pn=50001"
- "--rc=opc.tcp://host.docker.internal:4844"
- "--ut"
- "--aa"
- "--alm"
healthcheck:
test: ["CMD-SHELL", "netstat -an | grep -q ':50001.*LISTEN' || exit 1"]
interval: 5s
timeout: 2s
retries: 10
start_period: 10s

56
tests/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests/OpcPlcReverseConnectFixture.cs Normal file
View File

@@ -0,0 +1,56 @@
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests;
/// <summary>
/// Fixture for the reverse-connect variant of opc-plc (PR-11). Unlike the
/// dial-mode <see cref="OpcPlcFixture"/>, the simulator here is the dialer:
/// it reaches OUT to the test runner's listener URL on
/// <c>opc.tcp://host.docker.internal:4844</c>. The fixture's job is to
/// advertise the listener URL the test should bind and provide a clear
/// skip reason when the docker-compose service isn't running.
/// </summary>
/// <remarks>
/// <para>
/// <b>Why no port-probe</b>: the conventional fixture probes the simulator's
/// server port to detect docker-up. In reverse-connect the simulator opens
/// no inbound port — it's a pure dialer — so a probe would always fail.
/// Tests that want a hard skip should look at <see cref="SkipReason"/>
/// which is set from the <c>OPCUA_RC_SIM</c> env var (any value =
/// "simulator running"; absent = skip).
/// </para>
/// <para>
/// The "shared listener URL" model is enforced by the driver's
/// <c>ReverseConnectListener</c> singleton — multiple smoke tests in the
/// same xunit assembly share one listener instance even if they run in
/// parallel. Tests should pick distinct <c>ExpectedServerUri</c> values to
/// demultiplex inbound connections.
/// </para>
/// </remarks>
public sealed class OpcPlcReverseConnectFixture : IAsyncDisposable
{
private const string DefaultListenerUrl = "opc.tcp://0.0.0.0:4844";
private const string EnvVar = "OPCUA_RC_SIM";
/// <summary>The listener URL the driver should bind for incoming reverse-connect dials.</summary>
public string ListenerUrl { get; } = DefaultListenerUrl;
/// <summary>Skip reason when the reverse-connect simulator isn't available; null when ready.</summary>
public string? SkipReason { get; }
public OpcPlcReverseConnectFixture()
{
if (string.IsNullOrWhiteSpace(Environment.GetEnvironmentVariable(EnvVar)))
{
SkipReason =
$"Reverse-connect smoke skipped — set {EnvVar}=1 once `docker compose -f Docker/docker-compose.yml up opc-plc-rc` is healthy. " +
"The dialer needs host.docker.internal to reach this machine — verify Docker Desktop's network mode supports it.";
}
}
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
}
[Xunit.CollectionDefinition(Name)]
public sealed class OpcPlcReverseConnectCollection : Xunit.ICollectionFixture<OpcPlcReverseConnectFixture>
{
public const string Name = "OpcPlcReverseConnect";
}

65
tests/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests/OpcUaClientReverseConnectSmokeTests.cs Normal file
View File

@@ -0,0 +1,65 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests;
/// <summary>
/// Reverse-connect smoke (PR-11). Asserts the driver binds a listener at the
/// configured URL and accepts an inbound dial from <c>opc-plc-rc</c> (the
/// reverse-connect variant of Microsoft Industrial IoT's OPC UA simulator).
/// The session that comes up should be functionally identical to a dialled
/// session — same Read / Subscribe surface — but the transport direction is
/// server → client instead of client → server.
/// </summary>
/// <remarks>
/// <para>
/// <b>Build-only by default</b>: the test is gated on <c>OPCUA_RC_SIM</c>
/// + the docker-compose <c>opc-plc-rc</c> service. CI runs that don't
/// spin up the dialer skip with a clear message; the build still has to
/// compile so wire-level regressions in the reverse-connect code path are
/// caught even when the dialer isn't around.
/// </para>
/// </remarks>
[Collection(OpcPlcReverseConnectCollection.Name)]
[Trait("Category", "Integration")]
[Trait("Simulator", "opc-plc-rc")]
public sealed class OpcUaClientReverseConnectSmokeTests(OpcPlcReverseConnectFixture rc)
{
[Fact]
public async Task Driver_accepts_reverse_connect_from_opc_plc_rc_simulator()
{
if (rc.SkipReason is not null) Assert.Skip(rc.SkipReason);
var options = new OpcUaClientDriverOptions
{
// Conventional EndpointUrl still required — the driver derives the
// EndpointDescription from it for the session-create call. The actual
// dial direction is flipped by ReverseConnect.Enabled below.
EndpointUrl = "opc.tcp://opc-plc-rc:50001",
SecurityPolicy = OpcUaSecurityPolicy.None,
SecurityMode = OpcUaSecurityMode.None,
AuthType = OpcUaAuthType.Anonymous,
AutoAcceptCertificates = true,
Timeout = TimeSpan.FromSeconds(15),
SessionTimeout = TimeSpan.FromSeconds(60),
ReverseConnect = new ReverseConnectOptions(
Enabled: true,
ListenerUrl: rc.ListenerUrl,
// null = accept any upstream — only one is dialling this listener in
// the smoke test, so there's no demux to worry about.
ExpectedServerUri: null),
};
await using var drv = new OpcUaClientDriver(options, "opcua-rc-smoke");
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
// The session is up via reverse path — assert a steady-state read works.
var snapshots = await drv.ReadAsync(
[OpcPlcProfile.StepUp], TestContext.Current.CancellationToken);
snapshots.Count.ShouldBe(1);
snapshots[0].StatusCode.ShouldBe(0u,
"reverse-connect session must round-trip a Read identically to a dialled session");
}
}

259
tests/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReverseConnectTests.cs Normal file
View File

@@ -0,0 +1,259 @@
using System.Text.Json;
using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;
/// <summary>
/// Unit tests for the reverse-connect (server-initiated) path (PR-11). The driver
/// exposes two test seams — <c>ReverseConnectWaitHookForTest</c> and
/// <c>ReverseConnectSessionFactoryForTest</c> — that bypass the SDK's port-bind +
/// real-transport machinery so we can assert the wiring without standing up an
/// actual reverse-connect TCP listener.
/// </summary>
[Trait("Category", "Unit")]
public sealed class OpcUaClientReverseConnectTests : IDisposable
{
public void Dispose()
{
// Each test acquires/releases its own listener entries — but if a test fails
// mid-way the static dictionary in ReverseConnectListener could be left dirty.
// Nothing else here to do; the dictionary is internal and tests target unique
// listener URLs to avoid cross-test contamination.
}
[Fact]
public void ReverseConnect_options_default_to_disabled()
{
var opts = new OpcUaClientDriverOptions();
opts.ReverseConnect.ShouldNotBeNull();
opts.ReverseConnect.Enabled.ShouldBeFalse(
"default deployments dial outbound — reverse-connect is opt-in for OT-DMZ networks");
opts.ReverseConnect.ListenerUrl.ShouldBeNull();
opts.ReverseConnect.ExpectedServerUri.ShouldBeNull();
}
[Fact]
public async Task Disabled_reverse_connect_uses_existing_dial_path()
{
// Initialize against an unreachable endpoint with reverse-connect off; we expect
// the failover sweep to fail (since 192.0.2.x is reserved-for-documentation and
// routes nowhere). The point is to assert the dial path runs — not the listener
// path — by checking the wait hook was never invoked.
var waitInvoked = false;
var opts = new OpcUaClientDriverOptions
{
EndpointUrl = "opc.tcp://192.0.2.1:4840",
PerEndpointConnectTimeout = TimeSpan.FromMilliseconds(200),
Timeout = TimeSpan.FromMilliseconds(200),
ReverseConnect = new ReverseConnectOptions(Enabled: false),
};
await using var drv = new OpcUaClientDriver(opts, "opcua-rc-disabled");
drv.ReverseConnectWaitHookForTest = (_, _, _) =>
{
waitInvoked = true;
return Task.FromResult<ITransportWaitingConnection>(null!);
};
await Should.ThrowAsync<Exception>(() =>
drv.InitializeAsync("{}", TestContext.Current.CancellationToken));
waitInvoked.ShouldBeFalse(
"reverse-connect wait hook must not run when ReverseConnect.Enabled=false");
drv.ReverseListenerForTest.ShouldBeNull();
}
[Fact]
public async Task Enabled_reverse_connect_invokes_wait_with_expected_server_uri()
{
Uri? capturedListener = null;
string? capturedServerUri = null;
var opts = new OpcUaClientDriverOptions
{
EndpointUrl = "opc.tcp://upstream:4840",
ReverseConnect = new ReverseConnectOptions(
Enabled: true,
ListenerUrl: "opc.tcp://0.0.0.0:14844",
ExpectedServerUri: "urn:upstream-plc:server"),
Timeout = TimeSpan.FromMilliseconds(500),
};
await using var drv = new OpcUaClientDriver(opts, "opcua-rc-wait");
drv.ReverseConnectWaitHookForTest = (uri, serverUri, _) =>
{
capturedListener = uri;
capturedServerUri = serverUri;
// Return null — the session-factory hook below short-circuits before it
// gets used so a null connection can't blow up downstream.
return Task.FromResult<ITransportWaitingConnection>(new FakeTransportWaitingConnection());
};
// Halt the path immediately after WaitForServerAsync — the only thing we're
// asserting in this test is that wait runs with the right args. Throwing here
// means InitializeAsync transitions to Faulted but the assertions on the captured
// state still hold.
drv.ReverseConnectSessionFactoryForTest = (_, _, _, _, _) =>
throw new InvalidOperationException("test stops before session create");
await Should.ThrowAsync<Exception>(() =>
drv.InitializeAsync("{}", TestContext.Current.CancellationToken));
capturedListener.ShouldNotBeNull();
capturedListener!.ToString().ShouldContain("14844");
capturedServerUri.ShouldBe("urn:upstream-plc:server");
}
[Fact]
public async Task Enabled_reverse_connect_passes_connection_into_session_factory()
{
// Assert that the ITransportWaitingConnection returned from WaitForServerAsync
// flows verbatim into the session-create hook. This is the load-bearing wiring
// — a swap of variables at this seam would silently route the wrong connection
// into the SDK and the failure would surface only at session activation time.
var stubConnection = new FakeTransportWaitingConnection();
ITransportWaitingConnection? observedConnection = null;
var opts = new OpcUaClientDriverOptions
{
EndpointUrl = "opc.tcp://upstream:4840",
ReverseConnect = new ReverseConnectOptions(
Enabled: true,
ListenerUrl: "opc.tcp://0.0.0.0:14845"),
};
await using var drv = new OpcUaClientDriver(opts, "opcua-rc-conn");
drv.ReverseConnectWaitHookForTest = (_, _, _) =>
Task.FromResult<ITransportWaitingConnection>(stubConnection);
drv.ReverseConnectSessionFactoryForTest = (_, conn, _, _, _) =>
{
observedConnection = conn;
// Throw to short-circuit the post-create wiring (KeepAlive, diagnostics)
// — this test only cares about the connection plumbing.
throw new InvalidOperationException("test stops at factory");
};
await Should.ThrowAsync<Exception>(() =>
drv.InitializeAsync("{}", TestContext.Current.CancellationToken));
observedConnection.ShouldBeSameAs(stubConnection);
}
[Fact]
public async Task Enabled_reverse_connect_without_listener_url_fails_fast()
{
var opts = new OpcUaClientDriverOptions
{
EndpointUrl = "opc.tcp://upstream:4840",
ReverseConnect = new ReverseConnectOptions(Enabled: true, ListenerUrl: null),
};
await using var drv = new OpcUaClientDriver(opts, "opcua-rc-missing-listener");
var ex = await Should.ThrowAsync<Exception>(() =>
drv.InitializeAsync("{}", TestContext.Current.CancellationToken));
// The actual InvalidOperationException is wrapped in nothing — InitializeAsync
// re-throws after marking Faulted. Inspect the inner-most message either way.
var msg = ex is AggregateException agg ? string.Join("|", agg.InnerExceptions.Select(e => e.Message)) : ex.Message;
msg.ShouldContain("ListenerUrl");
drv.GetHealth().State.ShouldBe(DriverState.Faulted);
}
[Fact]
public void Listener_acquire_release_is_refcounted_per_url()
{
var url = "opc.tcp://0.0.0.0:14001";
var a = ReverseConnectListener.AcquireForTest(url);
var b = ReverseConnectListener.AcquireForTest(url);
// Same URL = same underlying entry. Two drivers sharing a listener must
// see the same SDK manager, otherwise we'd be double-binding the port.
a.ShouldBeSameAs(b);
a.RefCountForTest.ShouldBe(2);
a.Release();
b.RefCountForTest.ShouldBe(1, "first release drops one reference, listener stays alive");
b.Release();
ReverseConnectListener.InstanceCountForTest(url).ShouldBe(0,
"last release tears down the listener so the port can be re-bound");
}
[Fact]
public void Listener_distinct_urls_get_distinct_entries()
{
var a = ReverseConnectListener.AcquireForTest("opc.tcp://0.0.0.0:14010");
var b = ReverseConnectListener.AcquireForTest("opc.tcp://0.0.0.0:14011");
a.ShouldNotBeSameAs(b,
"different listener URLs must own independent SDK managers — sharing would conflate inbound connections from unrelated upstreams");
a.Release();
b.Release();
}
[Fact]
public async Task Disposal_releases_reverse_listener_reference()
{
var listenerUrl = "opc.tcp://0.0.0.0:14020";
// Pre-acquire a reference so we can observe the refcount drop when the driver
// shuts down. The driver uses a real Acquire (not the test seam) only when the
// wait hook is null — but we don't run init here, we just assert the listener
// hand-off model.
var pre = ReverseConnectListener.AcquireForTest(listenerUrl);
pre.RefCountForTest.ShouldBe(1);
// Simulate: driver came up, acquired listener, shut down → refcount returns
// to 1 (just our pre).
var drvHeld = ReverseConnectListener.AcquireForTest(listenerUrl);
drvHeld.RefCountForTest.ShouldBe(2);
drvHeld.Release();
pre.RefCountForTest.ShouldBe(1);
pre.Release();
ReverseConnectListener.InstanceCountForTest(listenerUrl).ShouldBe(0);
await Task.CompletedTask;
}
[Fact]
public void DTO_json_round_trip_preserves_reverse_connect_settings()
{
// The driver host deserializes OpcUaClientDriverOptions from JSON, so a missing
// System.Text.Json contract on a new section would silently lose the operator's
// config. Round-trip a populated ReverseConnectOptions and check it survives
// the JSON boundary.
var opts = new OpcUaClientDriverOptions
{
EndpointUrl = "opc.tcp://upstream:4840",
ReverseConnect = new ReverseConnectOptions(
Enabled: true,
ListenerUrl: "opc.tcp://0.0.0.0:4844",
ExpectedServerUri: "urn:plant:upstream-server"),
};
var json = JsonSerializer.Serialize(opts);
var roundTripped = JsonSerializer.Deserialize<OpcUaClientDriverOptions>(json);
roundTripped.ShouldNotBeNull();
roundTripped!.ReverseConnect.ShouldNotBeNull();
roundTripped.ReverseConnect.Enabled.ShouldBeTrue();
roundTripped.ReverseConnect.ListenerUrl.ShouldBe("opc.tcp://0.0.0.0:4844");
roundTripped.ReverseConnect.ExpectedServerUri.ShouldBe("urn:plant:upstream-server");
}
/// <summary>
/// Bare-bones <see cref="ITransportWaitingConnection"/> test double. The interface
/// surface is small (EndpointUrl + ServerUri + Handle) so a hand-rolled fake
/// keeps unit tests independent of any specific mocking library.
/// </summary>
private sealed class FakeTransportWaitingConnection : ITransportWaitingConnection
{
public Uri EndpointUrl { get; } = new("opc.tcp://fake-upstream:4840");
public string ServerUri => "urn:fake-upstream";
public object Handle => new object();
}
}