/* ======================================================================== * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved. * * OPC Foundation MIT License 1.00 * * Permission is hereby granted, free of charge, to any person * obtaining a copy of this software and associated documentation * files (the "Software"), to deal in the Software without * restriction, including without limitation the rights to use, * copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the * Software is furnished to do so, subject to the following * conditions: * * The above copyright notice and this permission notice shall be * included in all copies or substantial portions of the Software. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR * OTHER DEALINGS IN THE SOFTWARE. * * The complete license agreement can be found here: * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ using System; using System.Collections.Generic; using System.Diagnostics.CodeAnalysis; using System.IO; using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; using Opc.Ua.Security.Certificates; namespace Opc.Ua.Client { /// /// The implementation of a reverse connect client manager. /// /// /// This reverse connect manager allows to register for reverse connections /// with various strategies: /// i) take any connection. /// ii) filter for a specific application Uri and Url scheme. /// iii) filter for the Url. /// Second, any filter can be combined with the Once or Always flag. /// public class ReverseConnectManager : IDisposable { /// /// A default value for reverse hello configurations, if undefined. /// /// /// This value is used as wait timeout if the value is undefined by a caller. /// public const int DefaultWaitTimeout = 20000; /// /// Internal state of the reverse connect manager. /// private enum ReverseConnectManagerState { New = 0, Stopped = 1, Started = 2, Errored = 3 } /// /// Internal state of the reverse connect host. /// private enum ReverseConnectHostState { New = 0, Closed = 1, Open = 2, Errored = 3 } /// /// Specify the strategy for the reverse connect registration. /// [Flags] public enum ReverseConnectStrategy { /// /// Undefined strategy, defaults to Once. /// Undefined = 0, /// /// Remove entry after reverse connect callback. /// Once = 1, /// /// Always callback on matching url or uri. /// Always = 2, /// /// Flag for masking any connection. /// Any = 0x80, /// /// Respond to any incoming reverse connection, /// remove entry after reverse connect callback. /// AnyOnce = Any | Once, /// /// Respond to any incoming reverse connection, /// always callback. /// AnyAlways = Any | Always } /// /// Entry for a client reverse connect registration. /// private class ReverseConnectInfo { public ReverseConnectInfo(ReverseConnectHost reverseConnectHost, bool configEntry) { ReverseConnectHost = reverseConnectHost; State = ReverseConnectHostState.New; ConfigEntry = configEntry; } public ReverseConnectHost ReverseConnectHost; public ReverseConnectHostState State; public bool ConfigEntry; } /// /// Record to store information on a client /// registration for a reverse connect event. /// private class Registration { public Registration( string? serverUri, Uri endpointUrl, EventHandler onConnectionWaiting) : this(endpointUrl, onConnectionWaiting) { ServerUri = Utils.ReplaceLocalhost(serverUri); } /// /// Register with the server certificate to extract the application Uri. /// /// /// The first Uri in the subject alternate name field is considered the application Uri. /// /// The server certificate with the application Uri. /// The endpoint Url of the server. /// The connection to use. public Registration( Certificate serverCertificate, Uri endpointUrl, EventHandler onConnectionWaiting) : this(endpointUrl, onConnectionWaiting) { IReadOnlyList serverUris = X509Utils.GetApplicationUrisFromCertificate(serverCertificate); ServerUri = serverUris.Count != 0 ? serverUris[0] : null; } private Registration( Uri endpointUrl, EventHandler onConnectionWaiting) { EndpointUrl = new Uri(Utils.ReplaceLocalhost(endpointUrl.ToString())); OnConnectionWaiting = onConnectionWaiting; ReverseConnectStrategy = ReverseConnectStrategy.Once; } public readonly string? ServerUri; public readonly Uri EndpointUrl; public readonly EventHandler OnConnectionWaiting; public ReverseConnectStrategy ReverseConnectStrategy; } /// /// Obsolete default constructor /// [Obsolete("Use ReverseConnectManager(ITelemetryContext) instead.")] public ReverseConnectManager() // Forwards null into a non-nullable telemetry parameter on the modern ctor; // preserves the pre-nullable parameterless behavior of this obsolete API. : this(null!) { } /// /// Initializes the object with default values. /// public ReverseConnectManager(ITelemetryContext telemetry) : this(telemetry, timeProvider: null) { } /// /// Initializes the object with default values. /// /// The telemetry context to use to create obvservability instruments. /// Optional time provider used for elapsed-time calculations and /// async timeouts. Defaults to . public ReverseConnectManager(ITelemetryContext telemetry, TimeProvider? timeProvider) { m_timeProvider = timeProvider ?? TimeProvider.System; m_telemetry = telemetry; m_logger = telemetry.CreateLogger(); m_state = ReverseConnectManagerState.New; m_registrations = []; m_endpointUrls = []; m_cts = new CancellationTokenSource(); } /// /// Dispose implementation. /// public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } /// /// An overrideable version of the Dispose. /// /// true to release both managed and unmanaged resources; false to release only unmanaged resources. protected virtual void Dispose(bool disposing) { // close the watcher. m_configurationWatcher?.Dispose(); m_configurationWatcher = null; m_cts?.Dispose(); DisposeHosts(); } /// /// Raised when the configuration changes. /// /// The sender. /// The instance containing the event data. protected virtual async void OnConfigurationChangedAsync( object? sender, ConfigurationWatcherEventArgs args) { try { ApplicationConfiguration configuration = await ApplicationConfiguration .LoadAsync( new FileInfo(args.FilePath), m_applicationType, m_configType, m_telemetry) .ConfigureAwait(false); OnUpdateConfiguration(configuration); } catch (Exception e) { m_logger.LogError(e, "Could not load updated configuration file from: {FilePath}", args.FilePath); } } /// /// Called when the configuration is changed on disk. /// /// The configuration. [UnconditionalSuppressMessage("Trimming", "IL2074", Justification = "The configuration type was loaded with PublicParameterlessConstructor, so GetType() is safe to store.")] protected virtual void OnUpdateConfiguration(ApplicationConfiguration configuration) { // save types for config watcher m_applicationType = configuration.ApplicationType; m_configType = configuration.GetType(); // ClientConfiguration and ReverseConnect are nullable on ApplicationConfiguration, // but the file watcher is only enabled for client configurations that include a // populated ReverseConnect section, so both are guaranteed non-null here. OnUpdateConfiguration(configuration.ClientConfiguration!.ReverseConnect!); } /// /// Called when the reverse connect configuration is changed. /// /// /// An empty configuration or null stops service on all configured endpoints. /// /// The client endpoint configuration. protected virtual void OnUpdateConfiguration( ReverseConnectClientConfiguration configuration) { bool restartService = false; lock (m_lock) { if (m_configuration != null) { StopService(); m_configuration = null; restartService = true; } m_configuration = configuration ?? new ReverseConnectClientConfiguration(); // clear configured endpoints ClearEndpoints(true); if (configuration?.ClientEndpoints != null) { foreach (ReverseConnectClientEndpoint endpoint in configuration.ClientEndpoints) { Uri? uri = Utils.ParseUri(endpoint.EndpointUrl); if (uri != null) { AddEndpointInternal(uri, true); } } } if (restartService) { StartService(); } } } /// /// Open host ports. /// private void OpenHosts() { lock (m_lock) { foreach (KeyValuePair host in m_endpointUrls) { ReverseConnectInfo value = host.Value; try { if (host.Value.State < ReverseConnectHostState.Open) { value.ReverseConnectHost.Open(); value.State = ReverseConnectHostState.Open; } } catch (Exception e) { m_logger.LogError(e, "Failed to Open {Uri}.", host.Key); value.State = ReverseConnectHostState.Errored; } } } } /// /// Close host ports. /// private void CloseHosts() { lock (m_lock) { foreach (KeyValuePair host in m_endpointUrls) { ReverseConnectInfo value = host.Value; try { if (value.State == ReverseConnectHostState.Open) { value.ReverseConnectHost.Close(); value.State = ReverseConnectHostState.Closed; } } catch (Exception e) { m_logger.LogError(e, "Failed to Close {Uri}.", host.Key); value.State = ReverseConnectHostState.Errored; } } } } /// /// Dispose the hosts; /// private void DisposeHosts() { lock (m_lock) { CloseHosts(); m_endpointUrls.Clear(); } } /// /// Add endpoint for reverse connection. /// /// is null. /// public void AddEndpoint(Uri endpointUrl) { if (endpointUrl == null) { throw new ArgumentNullException(nameof(endpointUrl)); } lock (m_lock) { if (m_state == ReverseConnectManagerState.Started) { throw new ServiceResultException(StatusCodes.BadInvalidState); } AddEndpointInternal(endpointUrl, false); } } /// /// Starts the server application. /// /// The configuration. /// is null. /// public void StartService(ApplicationConfiguration configuration) { if (configuration == null) { throw new ArgumentNullException(nameof(configuration)); } lock (m_lock) { if (m_state == ReverseConnectManagerState.Started) { throw new ServiceResultException(StatusCodes.BadInvalidState); } try { OnUpdateConfiguration(configuration); StartService(); // monitor the configuration file. if (!string.IsNullOrEmpty(configuration.SourceFilePath)) { m_configurationWatcher = new ConfigurationWatcher(configuration, m_telemetry); m_configurationWatcher.Changed += OnConfigurationChangedAsync; } } catch (Exception e) { m_logger.LogError(e, "Unexpected error starting reverse connect manager."); m_state = ReverseConnectManagerState.Errored; var error = ServiceResult.Create( e, StatusCodes.BadInternalError, "Unexpected error starting application"); throw new ServiceResultException(error); } } } /// /// Starts the server application. /// /// The configuration. /// is null. /// public void StartService(ReverseConnectClientConfiguration configuration) { if (configuration == null) { throw new ArgumentNullException(nameof(configuration)); } lock (m_lock) { if (m_state == ReverseConnectManagerState.Started) { throw new ServiceResultException(StatusCodes.BadInvalidState); } try { m_configurationWatcher = null; OnUpdateConfiguration(configuration); OpenHosts(); m_state = ReverseConnectManagerState.Started; } catch (Exception e) { m_logger.LogError(e, "Unexpected error starting reverse connect manager."); m_state = ReverseConnectManagerState.Errored; var error = ServiceResult.Create( e, StatusCodes.BadInternalError, "Unexpected error starting reverse connect manager"); throw new ServiceResultException(error); } } } /// /// Clears all waiting reverse connection handlers. /// public void ClearWaitingConnections() { lock (m_registrationsLock) { m_registrations.Clear(); CancelAndRenewTokenSource(); } } /// /// Helper to wait for a reverse connection. /// [Obsolete("Use WaitForConnectionAsync instead.")] public Task WaitForConnection( Uri endpointUrl, string serverUri, CancellationToken ct = default) { return WaitForConnectionAsync(endpointUrl, serverUri, ct); } /// /// Helper to wait for a reverse connection. /// /// public async Task WaitForConnectionAsync( Uri endpointUrl, string? serverUri, CancellationToken ct = default) { var tcs = new TaskCompletionSource(); int hashCode = RegisterWaitingConnection( endpointUrl, serverUri, (sender, e) => tcs.TrySetResult(e), ReverseConnectStrategy.Once); async Task ListenForCancelAsync(CancellationToken ct) { if (ct == default) { int waitTimeout = m_configuration?.WaitTimeout ?? 20000; if (waitTimeout <= 0) { waitTimeout = DefaultWaitTimeout; } await Task.Delay(waitTimeout, ct).ConfigureAwait(false); } else { await Task.Delay(-1, ct).ContinueWith( _ => { }, ct, TaskContinuationOptions.None, TaskScheduler.Default).ConfigureAwait(false); } tcs.TrySetCanceled(ct); } await Task.WhenAny([tcs.Task, ListenForCancelAsync(ct)]).ConfigureAwait(false); if (!tcs.Task.IsCompleted || tcs.Task.IsCanceled) { UnregisterWaitingConnection(hashCode); throw new ServiceResultException( StatusCodes.BadTimeout, "Waiting for the reverse connection timed out."); } return await tcs.Task.ConfigureAwait(false); } /// /// Register for a waiting reverse connection. /// /// The endpoint Url of the reverse connection. /// Optional. The server application Uri of the reverse connection. /// The callback /// The reverse connect callback strategy. /// is null. public int RegisterWaitingConnection( Uri endpointUrl, string? serverUri, EventHandler onConnectionWaiting, ReverseConnectStrategy reverseConnectStrategy) { if (endpointUrl == null) { throw new ArgumentNullException(nameof(endpointUrl)); } var registration = new Registration(serverUri, endpointUrl, onConnectionWaiting) { ReverseConnectStrategy = reverseConnectStrategy }; lock (m_registrationsLock) { m_registrations.Add(registration); CancelAndRenewTokenSource(); } return registration.GetHashCode(); } /// /// Unregister reverse connection callback. /// /// The hashcode returned by the registration. public void UnregisterWaitingConnection(int hashCode) { lock (m_registrationsLock) { Registration? toRemove = null; foreach (Registration registration in m_registrations) { if (registration.GetHashCode() == hashCode) { toRemove = registration; break; } } if (toRemove != null) { m_registrations.Remove(toRemove); CancelAndRenewTokenSource(); } } } /// /// Called before the server stops /// private void StopService() { ClearWaitingConnections(); lock (m_lock) { CloseHosts(); m_state = ReverseConnectManagerState.Stopped; } } /// /// Called to start hosting the reverse connect ports. /// private void StartService() { lock (m_lock) { OpenHosts(); m_state = ReverseConnectManagerState.Started; } } /// /// Remove configuration endpoints from list. /// private void ClearEndpoints(bool configEntry) { var newEndpointUrls = new Dictionary(); foreach (KeyValuePair endpoint in m_endpointUrls) { if (endpoint.Value.ConfigEntry != configEntry) { newEndpointUrls[endpoint.Key] = endpoint.Value; } } m_endpointUrls = newEndpointUrls; } /// /// Add endpoint for reverse connection. /// /// The endpoint Url of the reverse connect client endpoint. /// Tf this is an entry in the application configuration. private void AddEndpointInternal(Uri endpointUrl, bool configEntry) { var reverseConnectHost = new ReverseConnectHost(m_telemetry); var info = new ReverseConnectInfo(reverseConnectHost, configEntry); try { m_endpointUrls[endpointUrl] = info; reverseConnectHost.CreateListener( endpointUrl, new ConnectionWaitingHandlerAsync(OnConnectionWaitingAsync), new EventHandler(OnConnectionStatusChanged)); } catch (ArgumentException ae) { m_logger.LogError(ae, "No listener was found for endpoint {EndpointUrl}.", endpointUrl); info.State = ReverseConnectHostState.Errored; } } /// /// Raised when a reverse connection is waiting, /// finds and calls a waiting connection. /// private async Task OnConnectionWaitingAsync(object sender, ConnectionWaitingEventArgs e) { long startTimestamp = m_timeProvider.GetTimestamp(); var holdTime = TimeSpan.FromMilliseconds(m_configuration?.HoldTime ?? 15000); bool matched = MatchRegistration(sender, e); while (!matched) { m_logger.LogInformation("Holding reverse connection: {ServerUri} {EndpointUrl}", e.ServerUri, e.EndpointUrl); CancellationToken ct; lock (m_registrationsLock) { ct = m_cts.Token; } TimeSpan delay = holdTime - m_timeProvider.GetElapsedTime(startTimestamp); if (delay > TimeSpan.Zero) { await m_timeProvider.Delay(delay, ct) .ContinueWith(tsk => { if (tsk.IsCanceled) { matched = MatchRegistration(sender, e); if (matched) { m_logger.LogInformation( "Matched reverse connection {ServerUri} {EndpointUrl} after {Duration}ms", e.ServerUri, e.EndpointUrl, (long)m_timeProvider.GetElapsedTime(startTimestamp).TotalMilliseconds); } } }, default, TaskContinuationOptions.None, TaskScheduler.Default) .ConfigureAwait(false); } break; } m_logger.LogInformation( "{Action} reverse connection: {ServerUri} {EndpointUrl} after {Duration}ms", e.Accepted ? "Accepted" : "Rejected", e.ServerUri, e.EndpointUrl, (long)m_timeProvider.GetElapsedTime(startTimestamp).TotalMilliseconds); } /// /// Match the waiting connection with a registration, callback registration, /// return if connection is accepted in event. /// /// true if a match was found. private bool MatchRegistration(object sender, ConnectionWaitingEventArgs e) { Registration? callbackRegistration = null; bool found = false; lock (m_registrationsLock) { // first try to match single registrations foreach (Registration registration in m_registrations .Where(r => (r.ReverseConnectStrategy & ReverseConnectStrategy.Any) == 0)) { if (registration.EndpointUrl.Scheme .Equals(e.EndpointUrl.Scheme, StringComparison.Ordinal) && (registration.ServerUri? .Equals(e.ServerUri, StringComparison.Ordinal) == true || registration.EndpointUrl.Authority.Equals(e.EndpointUrl.Authority, StringComparison.OrdinalIgnoreCase))) { callbackRegistration = registration; e.Accepted = true; found = true; m_logger.LogInformation( "Accepted reverse connection: {ServerUri} {EndpointUrl}", e.ServerUri, e.EndpointUrl); break; } } // now try any registrations. if (callbackRegistration == null) { foreach (Registration registration in m_registrations.Where(r => (r.ReverseConnectStrategy & ReverseConnectStrategy.Any) != 0)) { if (registration.EndpointUrl.Scheme .Equals(e.EndpointUrl.Scheme, StringComparison.Ordinal)) { callbackRegistration = registration; e.Accepted = true; found = true; m_logger.LogInformation( "Accept any reverse connection for approval: {ServerUri} {EndpointUrl}", e.ServerUri, e.EndpointUrl); break; } } } if (callbackRegistration != null && (callbackRegistration.ReverseConnectStrategy & ReverseConnectStrategy.Once) != 0) { m_registrations.Remove(callbackRegistration); } } callbackRegistration?.OnConnectionWaiting?.Invoke(sender, e); return found; } /// /// Raised when a connection status changes. /// private void OnConnectionStatusChanged(object? sender, ConnectionStatusEventArgs e) { m_logger.LogInformation( "Channel status: {EndpointUrl} {ChannelStatus} {Closed}", e.EndpointUrl, e.ChannelStatus, e.Closed); } /// /// Renew the cancellation token after use. /// private void CancelAndRenewTokenSource() { CancellationTokenSource cts = m_cts; m_cts = new CancellationTokenSource(); cts.Cancel(); cts.Dispose(); } [DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] private Type? m_configType; private readonly Lock m_lock = new(); private readonly TimeProvider m_timeProvider; private readonly ILogger m_logger; private readonly ITelemetryContext m_telemetry; private ConfigurationWatcher? m_configurationWatcher; private ApplicationType m_applicationType; private ReverseConnectClientConfiguration? m_configuration; private Dictionary m_endpointUrls; private ReverseConnectManagerState m_state; private readonly List m_registrations; private readonly Lock m_registrationsLock = new(); private CancellationTokenSource m_cts; } }