From 3af78d77114afea65c0b7cf0bf6c9cb3d885375c Mon Sep 17 00:00:00 2001 From: Suciu Mircea Adrian Date: Sat, 12 Sep 2026 08:30:57 +0300 Subject: [PATCH 1/6] Remove unused fields from ReferenceNodeManager (#4472) # Description Remove unused fields from ReferenceNodeManager --- .../ReferenceServer/ReferenceNodeManager.cs | 20 ------------------- 1 file changed, 20 deletions(-) diff --git a/Applications/Quickstarts.Servers/ReferenceServer/ReferenceNodeManager.cs b/Applications/Quickstarts.Servers/ReferenceServer/ReferenceNodeManager.cs index 7a0973c52e..3b113f48fb 100644 --- a/Applications/Quickstarts.Servers/ReferenceServer/ReferenceNodeManager.cs +++ b/Applications/Quickstarts.Servers/ReferenceServer/ReferenceNodeManager.cs @@ -5414,9 +5414,6 @@ protected override NodeState ValidateNode( "Aggregates_String" ]; - private static readonly bool[] s_booleanArray - = [true, false, true, false, true, false, true, false, true]; - private static readonly double[] s_doubleArray = [ 9.00001d, @@ -5435,23 +5432,6 @@ private static readonly float[] s_singleArray private static readonly int[] s_int32Array = [10, 11, 12, 13, 14, 15, 16, 17, 18, 19]; - private static readonly string[] s_stringArray - = ["en", "fr", "de", "en", "fr", "de", "en", "fr", "de", "en"]; - - private static readonly string[] s_stringArray0 = - [ - "a00", - "b10", - "c20", - "d30", - "e40", - "f50", - "g60", - "h70", - "i80", - "j90" - ]; - private static readonly string[] s_stringArray1 = ["open", "closed", "jammed"]; private static readonly string[] s_stringArray2 = ["red", "green", "blue", "cyan"]; private static readonly string[] s_stringArray3 = ["lolo", "lo", "normal", "hi", "hihi"]; From 077d67f5ae6606f4f7bc591460b8a72710659c58 Mon Sep 17 00:00:00 2001 From: Suciu Mircea Adrian Date: Sat, 26 Sep 2026 21:15:08 +0300 Subject: [PATCH 2/6] Master378 fix incomplete message buffering (#4550) # Description Backport the essential incomplete-message memory protection from 57c0a32ed without changing public/protected APIs or configuration schemas. - Introduce a server-wide budget for retained intermediate chunks, with reliable buffer cleanup on completion, rejection, and channel shutdown. - Restrict channels without activated sessions to a 50% total-usage threshold. Track live session membership so full-budget eligibility follows activation, transfer, closure, and expiration cleanup. _Put an `x` in the boxes that apply. You can complete these step by step after opening the PR._ - [x] I have signed the [CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf) and read the [CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md) doc. - [x] I have added tests that prove my fix is effective or that my feature works and increased code coverage. - [ ] I have added all necessary documentation. - [ ] I have verified that my changes do not introduce (new) build or analyzer warnings. - [ ] I ran **all** tests locally using the **UA.slnx** solution against at least .net **framework** and .net **10**, and all passed. - [ ] I fixed **all** failing and flaky tests in the CI pipelines and **all** CodeQL warnings. - [ ] I have addressed **all** PR feedback received. --------- Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../Opc.Ua.Server/Server/StandardServer.cs | 34 + .../Opc.Ua.Server/Session/SessionManager.cs | 36 + Stack/Opc.Ua.Core/Opc.Ua.Core.csproj | 1 + Stack/Opc.Ua.Core/Stack/Server/ServerBase.cs | 10 + Stack/Opc.Ua.Core/Stack/Tcp/ChannelQuotas.cs | 4 + .../Stack/Tcp/ChunkReassemblyBudget.cs | 110 +++ .../Stack/Tcp/TcpListenerChannel.cs | 14 +- .../Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs | 116 ++- .../Stack/Tcp/TcpTransportListener.cs | 4 + .../Stack/Tcp/UaSCBinaryChannel.cs | 157 +++- .../Transport/TransportListenerSettings.cs | 4 + .../Stack/Transport/ChunkReassemblyTests.cs | 719 ++++++++++++++++++ .../SessionlessBudgetTests.cs | 322 ++++++++ 13 files changed, 1469 insertions(+), 62 deletions(-) create mode 100644 Stack/Opc.Ua.Core/Stack/Tcp/ChunkReassemblyBudget.cs create mode 100644 Tests/Opc.Ua.Core.Tests/Stack/Transport/ChunkReassemblyTests.cs create mode 100644 Tests/Opc.Ua.Server.Tests/SessionlessBudgetTests.cs diff --git a/Libraries/Opc.Ua.Server/Server/StandardServer.cs b/Libraries/Opc.Ua.Server/Server/StandardServer.cs index 1154597c0c..bb8e496aa7 100644 --- a/Libraries/Opc.Ua.Server/Server/StandardServer.cs +++ b/Libraries/Opc.Ua.Server/Server/StandardServer.cs @@ -52,6 +52,7 @@ public class StandardServer : SessionServerBase, IStandardServer /// resources. protected override void Dispose(bool disposing) { + Volatile.Write(ref m_reassemblySessionManager, null); if (disposing) { if (m_conformanceUnitsManager != null) @@ -2920,6 +2921,11 @@ protected override void OnServerStarting(ApplicationConfiguration configuration) { base.OnServerStarting(configuration); + // Listeners open before the session manager exists. Keep their callback + // stable and resolve the current manager at each lookup, including restarts. + Volatile.Write(ref m_reassemblySessionManager, null); + HasActivatedSession = ChannelHasActivatedSession; + // save minimum nonce length. m_minNonceLength = configuration.SecurityConfiguration.NonceLength; @@ -3158,6 +3164,7 @@ await subscriptionManager.StartupAsync(cancellationToken) // add the session manager to the datastore. m_serverInternal.SetSessionManager(sessionManager, subscriptionManager); + Volatile.Write(ref m_reassemblySessionManager, sessionManager); ServerError = null; @@ -3242,6 +3249,7 @@ await subscriptionManager.StartupAsync(cancellationToken) } catch (Exception e) { + Volatile.Write(ref m_reassemblySessionManager, null); const string message = "Unexpected error starting application"; m_logger.LogCritical(Utils.TraceMasks.StartStop, e, message); Utils.SilentDispose(m_serverInternal); @@ -3278,6 +3286,7 @@ await subscriptionManager.StartupAsync(cancellationToken) /// protected override async ValueTask OnServerStoppingAsync(CancellationToken cancellationToken = default) { + Volatile.Write(ref m_reassemblySessionManager, null); m_logger.LogInformation(Utils.TraceMasks.StartStop, "Server - Stopping."); ShutDownDelay(); @@ -3862,11 +3871,36 @@ private void SessionChannelKeepAliveEvent(ISession session, SessionEventReason r } } + private bool ChannelHasActivatedSession(string channelId) + { + ISessionManager manager = Volatile.Read(ref m_reassemblySessionManager); + if (manager == null || string.IsNullOrEmpty(channelId)) + { + return false; + } + if (manager.GetType() == typeof(SessionManager)) + { + return ((SessionManager)manager).HasActivatedSession(channelId); + } + + // Custom managers (including overrides of SessionManager's lifecycle) + // keep their existing contract and do not need to maintain our snapshot. + foreach (ISession session in manager.GetSessions()) + { + if (session.Activated && string.Equals(session.SecureChannelId, channelId, StringComparison.Ordinal)) + { + return true; + } + } + return false; + } + private OperationLimitsState OperationLimits => ServerInternal.ServerObject.ServerCapabilities.OperationLimits; private readonly Lock m_registrationLock = new(); private readonly SemaphoreSlim m_semaphoreSlim = new(1, 1); + private ISessionManager m_reassemblySessionManager; private IServerInternal m_serverInternal; private ConfigurationWatcher m_configurationWatcher; private ConfiguredEndpointCollection m_registrationEndpoints; diff --git a/Libraries/Opc.Ua.Server/Session/SessionManager.cs b/Libraries/Opc.Ua.Server/Session/SessionManager.cs index 538a9802d5..5a482f7b40 100644 --- a/Libraries/Opc.Ua.Server/Session/SessionManager.cs +++ b/Libraries/Opc.Ua.Server/Session/SessionManager.cs @@ -90,6 +90,7 @@ protected virtual void Dispose(bool disposing) // create snapshot of all sessions KeyValuePair[] sessions = [.. m_sessions]; m_sessions.Clear(); + RefreshActivatedChannels(); foreach (KeyValuePair sessionKeyValue in sessions) { @@ -136,6 +137,7 @@ public virtual void Shutdown() // dispose of session objects using a snapshot. KeyValuePair[] sessions = [.. m_sessions]; m_sessions.Clear(); + RefreshActivatedChannels(); foreach (KeyValuePair sessionKeyValue in sessions) { @@ -403,6 +405,9 @@ public virtual async ValueTask CreateSessionAsync( localeIds, serverNonceObject); + // Channel transfers need a refresh even when the identity context is unchanged. + RefreshActivatedChannels(); + // raise session related event. if (contextChanged) { @@ -440,6 +445,8 @@ public virtual void CloseSession(NodeId sessionId) // close the session if removed. if (session != null) { + RefreshActivatedChannels(); + // raise session related event. RaiseSessionEvent(session, SessionEventReason.Closing); @@ -679,9 +686,38 @@ await m_server.CloseSessionAsync(null, session.Id, false) } } + internal bool HasActivatedSession(string channelId) + { + return channelId != null && Volatile.Read(ref m_activatedChannels).Contains(channelId); + } + + private void RefreshActivatedChannels() + { + // Serialize the scan as well as publication so a delayed refresh cannot + // overwrite newer membership. Always scan the live registry, not event arguments. + lock (m_activatedChannelsLock) + { + var channels = new HashSet(StringComparer.Ordinal); + foreach (ISession session in m_sessions.Values) + { + string channelId = session.SecureChannelId; + if (session.Activated && !string.IsNullOrEmpty(channelId)) + { + channels.Add(channelId); + } + } + Volatile.Write(ref m_activatedChannels, channels); + } + } + private readonly SemaphoreSlim m_semaphoreSlim = new(1, 1); private readonly IServerInternal m_server; private readonly ILogger m_logger; + + // Published sets are never mutated. Readers do not take the refresh lock. + private readonly object m_activatedChannelsLock = new(); + private HashSet m_activatedChannels = new(StringComparer.Ordinal); + private readonly NodeIdDictionary m_sessions; private uint m_lastSessionId; private readonly ManualResetEvent m_shutdownEvent; diff --git a/Stack/Opc.Ua.Core/Opc.Ua.Core.csproj b/Stack/Opc.Ua.Core/Opc.Ua.Core.csproj index 5489da272c..7219f547b0 100644 --- a/Stack/Opc.Ua.Core/Opc.Ua.Core.csproj +++ b/Stack/Opc.Ua.Core/Opc.Ua.Core.csproj @@ -14,6 +14,7 @@ + $(PackageId).Debug diff --git a/Stack/Opc.Ua.Core/Stack/Server/ServerBase.cs b/Stack/Opc.Ua.Core/Stack/Server/ServerBase.cs index c97bbe3c0a..ad22dc933d 100644 --- a/Stack/Opc.Ua.Core/Stack/Server/ServerBase.cs +++ b/Stack/Opc.Ua.Core/Stack/Server/ServerBase.cs @@ -85,6 +85,7 @@ protected virtual void Dispose(bool disposing) TransportListeners.Clear(); } + m_chunkReassemblyBudget = null; // dispose any hosts. if (ServiceHosts != null) @@ -579,6 +580,7 @@ public virtual async ValueTask StopAsync(CancellationToken cancellationToken = d listeners.Clear(); } + m_chunkReassemblyBudget = null; // close the hosts. lock (ServiceHosts) @@ -847,8 +849,13 @@ public virtual void CreateServiceHostEndpoint( // create the stack listener. try { + // All TCP endpoints of this server share the retained-message limit. + m_chunkReassemblyBudget ??= Bindings.ChunkReassemblyBudget.CreateDefault( + endpointConfiguration?.MaxMessageSize ?? TcpMessageLimits.DefaultMaxMessageSize); var settings = new TransportListenerSettings { + ChunkReassemblyBudget = m_chunkReassemblyBudget, + HasActivatedSession = HasActivatedSession, Descriptions = endpoints, Configuration = endpointConfiguration, ServerCertificateTypesProvider = InstanceCertificateTypesProvider, @@ -1795,6 +1802,9 @@ await m_server.ProcessRequestAsync(request, ct) private RequestQueue m_requestQueue; private ITelemetryContext m_telemetry; + internal Func HasActivatedSession { get; set; } + + private Bindings.ChunkReassemblyBudget m_chunkReassemblyBudget; private bool m_disposed; } } diff --git a/Stack/Opc.Ua.Core/Stack/Tcp/ChannelQuotas.cs b/Stack/Opc.Ua.Core/Stack/Tcp/ChannelQuotas.cs index b536550681..2a5633258d 100644 --- a/Stack/Opc.Ua.Core/Stack/Tcp/ChannelQuotas.cs +++ b/Stack/Opc.Ua.Core/Stack/Tcp/ChannelQuotas.cs @@ -34,6 +34,10 @@ namespace Opc.Ua.Bindings /// public class ChannelQuotas { + internal ChunkReassemblyBudget ChunkReassemblyBudget { get; set; } + + internal System.Func HasActivatedSession { get; set; } + /// /// Creates an object with default values. /// diff --git a/Stack/Opc.Ua.Core/Stack/Tcp/ChunkReassemblyBudget.cs b/Stack/Opc.Ua.Core/Stack/Tcp/ChunkReassemblyBudget.cs new file mode 100644 index 0000000000..6845fd8fa3 --- /dev/null +++ b/Stack/Opc.Ua.Core/Stack/Tcp/ChunkReassemblyBudget.cs @@ -0,0 +1,110 @@ +/* ======================================================================== + * 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.Threading; + +namespace Opc.Ua.Bindings +{ + /// + /// Bounds backing arrays retained for incomplete requests across a server's channels. + /// + internal sealed class ChunkReassemblyBudget + { + internal ChunkReassemblyBudget(long maxBytes) + { + if (maxBytes <= 0) + { + throw new ArgumentOutOfRangeException(nameof(maxBytes)); + } + MaxBytes = maxBytes; + } + + internal long MaxBytes { get; } + + internal long ReservedBytes => Interlocked.Read(ref m_reservedBytes); + + internal static ChunkReassemblyBudget CreateDefault(int maxMessageSize) + { + const long min = 64L * 1024 * 1024; + const long max = 1024L * 1024 * 1024; + long capacity = maxMessageSize <= 0 ? max : + Math.Max(Math.Min(Math.Max(16L * maxMessageSize, min), max), 4L * maxMessageSize); + return new ChunkReassemblyBudget(capacity); + } + + internal bool TryReserve(long bytes, bool hasActivatedSession) + { + if (bytes < 0) + { + throw new ArgumentOutOfRangeException(nameof(bytes)); + } + // Sessionless reservations share the same total, but cannot cross half capacity. + long limit = hasActivatedSession ? MaxBytes : MaxBytes / 2; + long reserved = ReservedBytes; + while (true) + { + if (bytes > limit - reserved) + { + return false; + } + long observed = Interlocked.CompareExchange(ref m_reservedBytes, reserved + bytes, reserved); + if (observed == reserved) + { + return true; + } + reserved = observed; + } + } + + internal void Release(long bytes) + { + if (bytes < 0) + { + throw new ArgumentOutOfRangeException(nameof(bytes)); + } + long reserved = ReservedBytes; + while (true) + { + if (bytes > reserved) + { + throw new InvalidOperationException("More reassembly bytes released than reserved."); + } + long observed = Interlocked.CompareExchange(ref m_reservedBytes, reserved - bytes, reserved); + if (observed == reserved) + { + return; + } + reserved = observed; + } + } + + private long m_reservedBytes; + } +} diff --git a/Stack/Opc.Ua.Core/Stack/Tcp/TcpListenerChannel.cs b/Stack/Opc.Ua.Core/Stack/Tcp/TcpListenerChannel.cs index eed9edfed4..6394033c6b 100644 --- a/Stack/Opc.Ua.Core/Stack/Tcp/TcpListenerChannel.cs +++ b/Stack/Opc.Ua.Core/Stack/Tcp/TcpListenerChannel.cs @@ -196,9 +196,16 @@ public void IdleCleanup() public int ElapsedSinceLastActiveTime => HiResClock.TickCount - LastActiveTickCount; /// - /// Has the channel been used in a session + /// Whether the channel has an activated session, or the legacy activation hint + /// when no session membership provider is available. /// - public bool UsedBySession { get; protected set; } + public bool UsedBySession + { + get => Quotas.HasActivatedSession?.Invoke(GlobalChannelId) ?? m_usedBySession; + protected set => m_usedBySession = value; + } + + private bool m_usedBySession; /// /// Handles a socket error. @@ -257,6 +264,7 @@ protected void ForceChannelFault(ServiceResult reason) { lock (DataLock) { + TakeSavedChunks().Release(BufferManager, nameof(ForceChannelFault)); CompleteReverseHello(new ServiceResultException(reason)); // nothing to do if channel already in a faulted state. @@ -362,6 +370,7 @@ protected void ChannelClosed() finally { State = TcpChannelState.Closed; + ClosePartialMessage(); Listener.ChannelClosed(ChannelId); // notify any monitors. @@ -382,6 +391,7 @@ protected void ChannelFaulted() finally { State = TcpChannelState.Faulted; + ClosePartialMessage(); Listener.ChannelClosed(ChannelId); } } diff --git a/Stack/Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs b/Stack/Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs index afa60d7be4..e8480f0201 100644 --- a/Stack/Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs +++ b/Stack/Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs @@ -301,6 +301,10 @@ protected override bool HandleIncomingMessage( { lock (DataLock) { + if (State is TcpChannelState.Closed or TcpChannelState.Faulted) + { + return false; + } SetResponseRequired(true); try @@ -657,6 +661,10 @@ const string errorSecurityChecksFailed } // get the chunks to process. chunksToProcess = GetSavedChunks(requestId, messageBody, true); + if (State == TcpChannelState.Closed) + { + return false; + } request = (OpenSecureChannelRequest) BinaryDecoder.DecodeMessage( @@ -959,7 +967,7 @@ private bool ProcessCloseSecureChannelRequest( if (!TcpMessageType.IsFinal(messageType)) { SaveIntermediateChunk(requestId, messageBody, true); - return false; + return true; } // get the chunks to process. @@ -1094,41 +1102,39 @@ private bool ProcessRequestMessage(uint messageType, ArraySegment messageC "ChannelId {Id}: ProcessRequestMessage RequestId {RequestId} was aborted.", ChannelId, requestId); - chunksToProcess = GetSavedChunks(requestId, messageBody, true); + chunksToProcess = TakeSavedChunks(); + chunksToProcess.Add(messageBody); return true; } // check if it is necessary to wait for more chunks. if (!TcpMessageType.IsFinal(messageType)) { - bool firstChunk = SaveIntermediateChunk(requestId, messageBody, true); + // Validate before transferring ownership: a rejected chunk is + // returned to the pool immediately by SaveIntermediateChunk. + if (DiscoveryOnly && !HasPartialMessage && + !ValidateDiscoveryServiceCall(token, requestId, messageBody, out chunksToProcess)) + { + ChannelClosed(); + return true; + } - // validate the type is allowed with a discovery channel - if (DiscoveryOnly) + bool firstChunk = SaveIntermediateChunk(requestId, messageBody, true); + if (State == TcpChannelState.Closed) { - if (firstChunk) - { - if (!ValidateDiscoveryServiceCall( - token, - requestId, - messageBody, - out chunksToProcess)) - { - ChannelClosed(); - } - } - else if (GetSavedChunksTotalSize() > TcpMessageLimits - .DefaultDiscoveryMaxMessageSize) - { - chunksToProcess = GetSavedChunks(0, messageBody, true); - SendServiceFault( - token, - requestId, - ServiceResult.Create( - StatusCodes.BadSecurityPolicyRejected, - "Discovery Channel message size exceeded.")); - ChannelClosed(); - } + return true; + } + if (DiscoveryOnly && !firstChunk && + GetSavedChunksTotalSize() > TcpMessageLimits.DefaultDiscoveryMaxMessageSize) + { + chunksToProcess = TakeSavedChunks(); + SendServiceFault( + token, + requestId, + ServiceResult.Create( + StatusCodes.BadSecurityPolicyRejected, + "Discovery Channel message size exceeded.")); + ChannelClosed(); } return true; @@ -1152,6 +1158,10 @@ private bool ProcessRequestMessage(uint messageType, ArraySegment messageC // get the chunks to process. chunksToProcess = GetSavedChunks(requestId, messageBody, true); + if (State == TcpChannelState.Closed) + { + return true; + } // decode the request. if (BinaryDecoder.DecodeMessage( @@ -1310,6 +1320,36 @@ protected override void DoMessageLimitsExceeded() ChannelClosed(); } + internal override void OnReassemblyBudgetExceeded() + { + try + { + if (Socket != null) + { + SendErrorMessage(new ServiceResult( + StatusCodes.BadTcpNotEnoughResources, + "The server cannot retain more chunks of incomplete messages.")); + } + } + catch (Exception e) + { + m_logger.LogDebug(e, "Could not report exhausted reassembly capacity."); + } + finally + { + // The incoming buffer has already been returned. Do not let a + // failed terminal write/close make the receive loop return it twice. + try + { + ChannelClosed(); + } + catch (Exception e) + { + m_logger.LogDebug(e, "Error closing a channel after reassembly exhaustion."); + } + } + } + /// /// Validate the type of message before it is decoded. /// @@ -1320,15 +1360,27 @@ private bool ValidateDiscoveryServiceCall( out BufferCollection chunksToProcess) { chunksToProcess = null; - using var decoder = new BinaryDecoder(messageBody, Quotas.MessageContext); - // read the type of the message before more chunks are processed. - NodeId typeId = decoder.ReadNodeId(null); + NodeId typeId; + try + { + using var decoder = new BinaryDecoder(messageBody, Quotas.MessageContext); + // read the type of the message before more chunks are processed. + typeId = decoder.ReadNodeId(null); + } + catch + { + // Validation runs before SaveIntermediateChunk takes ownership. + // Let ProcessRequestMessage release this buffer after reporting the fault. + chunksToProcess = new BufferCollection(messageBody); + throw; + } if (typeId != ObjectIds.GetEndpointsRequest_Encoding_DefaultBinary && typeId != ObjectIds.FindServersRequest_Encoding_DefaultBinary && typeId != ObjectIds.FindServersOnNetworkRequest_Encoding_DefaultBinary) { - chunksToProcess = GetSavedChunks(0, messageBody, true); + chunksToProcess = TakeSavedChunks(); + chunksToProcess.Add(messageBody); SendServiceFault( token, requestId, diff --git a/Stack/Opc.Ua.Core/Stack/Tcp/TcpTransportListener.cs b/Stack/Opc.Ua.Core/Stack/Tcp/TcpTransportListener.cs index d25a6c2a22..ec8e7250b7 100644 --- a/Stack/Opc.Ua.Core/Stack/Tcp/TcpTransportListener.cs +++ b/Stack/Opc.Ua.Core/Stack/Tcp/TcpTransportListener.cs @@ -377,6 +377,10 @@ public void Open( m_quotas.CertificateValidator = settings.CertificateValidator; + m_quotas.HasActivatedSession = settings.HasActivatedSession; + m_quotas.ChunkReassemblyBudget = settings.ChunkReassemblyBudget ?? + ChunkReassemblyBudget.CreateDefault(m_quotas.MaxMessageSize); + // save the server certificate. m_serverCertificateTypesProvider = settings.ServerCertificateTypesProvider; diff --git a/Stack/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs b/Stack/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs index 7b98983001..6a7aa10ea3 100644 --- a/Stack/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs +++ b/Stack/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs @@ -225,6 +225,7 @@ protected virtual void Dispose(bool disposing) { if (disposing) { + ClosePartialMessage(); Socket?.Close(); DiscardTokens(); Utils.SilentDispose(Socket); @@ -382,42 +383,86 @@ protected bool SaveIntermediateChunk( ArraySegment chunk, bool isServerContext) { - bool firstChunk = false; - if (m_partialMessageChunks == null) + return SaveChunk(requestId, chunk, isServerContext, false, out _); + } + + // Takes ownership of chunk, including when it cannot be retained. + // Channel callbacks run after leaving the partial-message lock. + private bool SaveChunk( + uint requestId, + ArraySegment chunk, + bool isServerContext, + bool final, + out BufferCollection completed) + { + completed = null; + bool firstChunk; + bool limitsExceeded; + bool budgetExceeded = false; + lock (m_partialMessageLock) { - firstChunk = true; - m_partialMessageChunks = []; - } + if (m_partialMessageClosed) + { + BufferManager.ReturnBuffer(chunk.Array, nameof(SaveChunk)); + completed = []; + return false; + } - bool chunkOrSizeLimitsExceeded = MessageLimitsExceeded( - isServerContext, - m_partialMessageChunks.TotalSize, - m_partialMessageChunks.Count); + firstChunk = m_partialMessageChunks == null; + // Keep the existing negotiated message-limit policy. + limitsExceeded = MessageLimitsExceeded( + isServerContext, + m_partialMessageChunks?.TotalSize ?? 0, + m_partialMessageChunks?.Count ?? 0); + if (m_partialRequestId != requestId || limitsExceeded) + { + if (m_partialMessageChunks?.Count > 0) + { + m_logger.LogWarning( + "WARNING - Discarding unprocessed message chunks for Request #{PartialRequestId}", + m_partialRequestId); + } + ReleaseSavedChunks(); + } - if ((m_partialRequestId != requestId) || chunkOrSizeLimitsExceeded) - { - if (m_partialMessageChunks.Count > 0) + if (!limitsExceeded && requestId != 0 && chunk.Array != null) + { + ChunkReassemblyBudget budget = isServerContext && !final + ? Quotas.ChunkReassemblyBudget : null; + if (budget == null || budget.TryReserve( + chunk.Array.Length, this is TcpListenerChannel listener && listener.UsedBySession)) + { + m_partialMessageReservedBytes += budget == null ? 0 : chunk.Array.Length; + m_partialRequestId = requestId; + m_partialMessageChunks ??= []; + m_partialMessageChunks.Add(chunk); + } + else + { + ReleaseSavedChunks(); + BufferManager.ReturnBuffer(chunk.Array, nameof(SaveChunk)); + budgetExceeded = true; + } + } + else { - m_logger.LogWarning( - "WARNING - Discarding unprocessed message chunks for Request #{PartialRequestId}", - m_partialRequestId); + BufferManager.ReturnBuffer(chunk.Array, nameof(SaveChunk)); } - m_partialMessageChunks.Release(BufferManager, "SaveIntermediateChunk"); + if (final) + { + completed = TakeSavedChunks(); + } } - if (chunkOrSizeLimitsExceeded) + if (budgetExceeded) { - DoMessageLimitsExceeded(); - return firstChunk; + OnReassemblyBudgetExceeded(); } - - if (requestId != 0) + else if (limitsExceeded) { - m_partialRequestId = requestId; - m_partialMessageChunks.Add(chunk); + DoMessageLimitsExceeded(); } - return firstChunk; } @@ -429,10 +474,60 @@ protected BufferCollection GetSavedChunks( ArraySegment chunk, bool isServerContext) { - SaveIntermediateChunk(requestId, chunk, isServerContext); - BufferCollection savedChunks = m_partialMessageChunks; + SaveChunk(requestId, chunk, isServerContext, true, out BufferCollection completed); + return completed; + } + + internal bool HasPartialMessage + { + get + { + lock (m_partialMessageLock) + { + return m_partialMessageChunks != null; + } + } + } + + internal BufferCollection TakeSavedChunks() + { + lock (m_partialMessageLock) + { + BufferCollection chunks = m_partialMessageChunks ?? []; + m_partialMessageChunks = null; + ReleasePartialMessageReservation(); + return chunks; + } + } + + internal void ClosePartialMessage() + { + lock (m_partialMessageLock) + { + m_partialMessageClosed = true; + ReleaseSavedChunks(); + } + } + + private void ReleaseSavedChunks() + { + m_partialMessageChunks?.Release(BufferManager, nameof(ReleaseSavedChunks)); m_partialMessageChunks = null; - return savedChunks; + ReleasePartialMessageReservation(); + } + + private void ReleasePartialMessageReservation() + { + if (m_partialMessageReservedBytes != 0) + { + Quotas.ChunkReassemblyBudget.Release(m_partialMessageReservedBytes); + m_partialMessageReservedBytes = 0; + } + } + + internal virtual void OnReassemblyBudgetExceeded() + { + DoMessageLimitsExceeded(); } /// @@ -440,7 +535,10 @@ protected BufferCollection GetSavedChunks( /// protected int GetSavedChunksTotalSize() { - return m_partialMessageChunks?.TotalSize ?? 0; + lock (m_partialMessageLock) + { + return m_partialMessageChunks?.TotalSize ?? 0; + } } /// @@ -973,6 +1071,9 @@ public void UpdateLastActiveTime() private bool m_firstReceivedSequenceNumber = true; private uint m_partialRequestId; private BufferCollection m_partialMessageChunks; + private readonly object m_partialMessageLock = new(); + private long m_partialMessageReservedBytes; + private bool m_partialMessageClosed; private TcpChannelStateEventHandler m_stateChanged; private const uint kMaxValueLegacyTrue = TcpMessageLimits.MinSequenceNumber; diff --git a/Stack/Opc.Ua.Core/Stack/Transport/TransportListenerSettings.cs b/Stack/Opc.Ua.Core/Stack/Transport/TransportListenerSettings.cs index 25351e3b3a..f270c66caa 100644 --- a/Stack/Opc.Ua.Core/Stack/Transport/TransportListenerSettings.cs +++ b/Stack/Opc.Ua.Core/Stack/Transport/TransportListenerSettings.cs @@ -36,6 +36,10 @@ namespace Opc.Ua /// public class TransportListenerSettings { + internal Bindings.ChunkReassemblyBudget ChunkReassemblyBudget { get; set; } + + internal System.Func HasActivatedSession { get; set; } + /// /// Gets or sets the descriptions for the endpoints supported by the listener. /// diff --git a/Tests/Opc.Ua.Core.Tests/Stack/Transport/ChunkReassemblyTests.cs b/Tests/Opc.Ua.Core.Tests/Stack/Transport/ChunkReassemblyTests.cs new file mode 100644 index 0000000000..0c4bfe4f25 --- /dev/null +++ b/Tests/Opc.Ua.Core.Tests/Stack/Transport/ChunkReassemblyTests.cs @@ -0,0 +1,719 @@ +/* ======================================================================== + * 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.Buffers; +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Reflection; +using System.Threading.Tasks; +using Moq; +using NUnit.Framework; +using Opc.Ua.Bindings; +using Opc.Ua.Tests; +using Opc.Ua.Security.Certificates; + +namespace Opc.Ua.Core.Tests.Stack.Transport +{ + [TestFixture] + [Category("ChunkReassembly")] + public class ChunkReassemblyTests + { + [TestCase(0, 1073741824L)] + [TestCase(-1, 1073741824L)] + [TestCase(2097152, 67108864L)] + [TestCase(8388608, 134217728L)] + [TestCase(134217728, 1073741824L)] + [TestCase(int.MaxValue, 8589934588L)] + public void DefaultBudgetMatchesUpstreamSizing(int size, long expected) + { + Assert.That(ChunkReassemblyBudget.CreateDefault(size).MaxBytes, Is.EqualTo(expected)); + } + + [TestCase(false)] + [TestCase(true)] + public void ConcurrentReservationsNeverExceedCapacityAndCanBeReused(bool activated) + { + var budget = new ChunkReassemblyBudget(1000); + long limit = activated ? 1000 : 500; + Parallel.For(0, 100, _ => + { + for (int i = 0; i < 100; i++) + { + if (budget.TryReserve(37, activated)) + { + Assert.That(budget.ReservedBytes, Is.InRange(37L, limit)); + budget.Release(37); + } + } + }); + Assert.That(budget.ReservedBytes, Is.Zero); + Assert.That(budget.TryReserve(limit, activated), Is.True); + Assert.That(budget.TryReserve(1, activated), Is.False); + Assert.Throws(() => budget.Release(1001)); + budget.Release(limit); + } + + [TestCase(1000)] + [TestCase(1001)] + public void SessionlessThresholdIncludesActivatedReservations(long capacity) + { + var budget = new ChunkReassemblyBudget(capacity); + long half = capacity / 2; + Assert.That(budget.TryReserve(half - 1, true), Is.True); + Assert.That(budget.TryReserve(2, false), Is.False); + Assert.That(budget.TryReserve(1, false), Is.True); + Assert.That(budget.TryReserve(1, false), Is.False); + Assert.That(budget.TryReserve(capacity - half, true), Is.True); + Assert.That(budget.ReservedBytes, Is.EqualTo(capacity)); + budget.Release(capacity); + Assert.That(budget.TryReserve(half, false), Is.True); + budget.Release(half); + } + + [Test] + public void LiveMembershipOverridesActivationResponseHintAndDowngradeReleasesBuffers() + { + var budget = new ChunkReassemblyBudget(1024); + bool activated = true; + var quotas = new ChannelQuotas(new ServiceMessageContext(NUnitTelemetryContext.Create())) + { + ChunkReassemblyBudget = budget, + HasActivatedSession = id => id == "test-1" && activated + }; + using var owner = new TestChannel(budget, quotas); + owner.SetSession(false); // A failed/late response cannot downgrade a live session. + Assert.That(owner.UsedBySession, Is.True); + for (int i = 0; i < 3; i++) + { + owner.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [], 1); + } + Assert.That(budget.ReservedBytes, Is.EqualTo(768)); + using var sessionless = new TestChannel(budget); + sessionless.SetSession(false); + sessionless.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [], 1); + Assert.That(sessionless.Closed, Is.True); + Assert.That(budget.ReservedBytes, Is.EqualTo(768)); + + activated = false; + owner.SetSession(true); // A successful but stale response cannot restore membership. + Assert.That(owner.UsedBySession, Is.False); + Assert.That(budget.ReservedBytes, Is.EqualTo(768), "Downgrade does not evict retained chunks."); + owner.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [], 1); + Assert.That(owner.Closed, Is.True); + Assert.That(owner.Pool.Outstanding, Is.Zero); + Assert.That(budget.ReservedBytes, Is.Zero); + using var next = new TestChannel(budget); + next.SetSession(false); + next.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [], 1); + Assert.That(next.Closed, Is.False); + } + + [Test] + public void DowngradedChannelCanStillCompleteAnAlreadyRetainedRequest() + { + var budget = new ChunkReassemblyBudget(256); + using var channel = new TestChannel(budget); + byte[] body = channel.EncodeRequest(); + int split = body.Length / 2; + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, body.Take(split).ToArray(), 1); + channel.SetSession(false); + channel.Receive(TcpMessageType.Message | TcpMessageType.Final, body.Skip(split).ToArray(), 1); + Assert.That(channel.Requests.Count, Is.EqualTo(1)); + Assert.That(channel.Pool.Outstanding, Is.Zero); + Assert.That(budget.ReservedBytes, Is.Zero); + } + + [TestCase(false)] + [TestCase(true)] + public void ManyChannelsCannotRetainMoreThanTheSharedBudget(bool activatedSession) + { + var budget = new ChunkReassemblyBudget(1024); + var channels = new List(); + try + { + for (int i = 0; i < 40; i++) + { + var channel = new TestChannel(budget); + channels.Add(channel); + channel.SetSession(activatedSession); + // Empty bodies still retain a complete backing array. + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [], 1); + Assert.That(budget.ReservedBytes, Is.LessThanOrEqualTo(budget.MaxBytes)); + Assert.That(channel.Requests, Is.Empty); + if (channel.Closed) + { + Assert.That(channel.Pool.Outstanding, Is.Zero); + Assert.That(channel.LastError, Is.EqualTo(StatusCodes.BadTcpNotEnoughResources)); + } + } + Assert.That(channels.Count(c => !c.Closed), Is.EqualTo(activatedSession ? 4 : 2)); + Assert.That(budget.ReservedBytes, Is.EqualTo(activatedSession ? 1024 : 512)); + } + finally + { + foreach (TestChannel channel in channels) + { + channel.Dispose(); + Assert.That(channel.Pool.Outstanding, Is.Zero); + } + } + Assert.That(budget.ReservedBytes, Is.Zero); + } + + [TestCase(false)] + [TestCase(true)] + public void ExhaustionReleasesOldAndIncomingBuffersEvenIfReportingFails(bool failReporting) + { + var budget = new ChunkReassemblyBudget(256); + using var channel = new TestChannel(budget); + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [1], 1); + channel.FailReporting = failReporting; + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [2], 1); + Assert.That(channel.Closed, Is.True); + Assert.That(channel.Pool.Outstanding, Is.Zero); + Assert.That(budget.ReservedBytes, Is.Zero); + // A receive callback already queued when the socket closed must not dispatch. + channel.Receive(TcpMessageType.Message | TcpMessageType.Final, channel.EncodeRequest(), 1); + Assert.That(channel.Requests, Is.Empty); + Assert.That(channel.Pool.Outstanding, Is.Zero); + } + + [Test] + public void CompletionReleasesCapacityAndDispatchesValidChunkedRequests() + { + var budget = new ChunkReassemblyBudget(256); + using var channel = new TestChannel(budget); + byte[] body = channel.EncodeRequest(); + for (uint requestId = 1; requestId <= 2; requestId++) + { + int split = body.Length / 2; + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, body.Take(split).ToArray(), requestId); + Assert.That(budget.ReservedBytes, Is.EqualTo(256)); + channel.Receive(TcpMessageType.Message | TcpMessageType.Final, body.Skip(split).ToArray(), requestId); + Assert.That(budget.ReservedBytes, Is.Zero); + Assert.That(channel.Pool.Outstanding, Is.Zero); + } + Assert.That(channel.Requests.Count, Is.EqualTo(2)); + Assert.That(channel.Requests.All(r => r is ReadRequest), Is.True); + Assert.That(channel.Closed, Is.False); + } + + [Test] + public void SingleChunkRequestWorksWhileAnotherChannelHoldsTheEntireBudget() + { + var budget = new ChunkReassemblyBudget(256); + using var holder = new TestChannel(budget); + using var other = new TestChannel(budget); + holder.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [], 1); + other.SetSession(false); + other.Receive(TcpMessageType.Message | TcpMessageType.Final, other.EncodeRequest(), 1); + Assert.That(other.Requests.Count, Is.EqualTo(1)); + Assert.That(other.Pool.Outstanding, Is.Zero); + Assert.That(budget.ReservedBytes, Is.EqualTo(256)); + } + + [Test] + public void AbortReleasesCapacityAndKeepsTheChannelUsable() + { + var budget = new ChunkReassemblyBudget(256); + using var channel = new TestChannel(budget); + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [], 1); + channel.Receive(TcpMessageType.Message | TcpMessageType.Abort, [], 1); + Assert.That(budget.ReservedBytes, Is.Zero); + Assert.That(channel.Pool.Outstanding, Is.Zero); + Assert.That(channel.Closed, Is.False); + channel.Receive(TcpMessageType.Message | TcpMessageType.Final, channel.EncodeRequest(), 2); + Assert.That(channel.Requests.Count, Is.EqualTo(1)); + } + + [Test] + public void ReplacementReleasesPreviousReservationBeforeReservingNewChunk() + { + var budget = new ChunkReassemblyBudget(256); + using var channel = new TestChannel(budget); + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [1], 1); + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [2], 2); + Assert.That(budget.ReservedBytes, Is.EqualTo(256)); + Assert.That(channel.Pool.Outstanding, Is.EqualTo(1)); + Assert.That(channel.Closed, Is.False); + } + + [TestCase("dispose")] + [TestCase("close")] + [TestCase("fault")] + [TestCase("disconnect")] + public void TerminalAndFaultPathsReleasePartialMessages(string action) + { + var budget = new ChunkReassemblyBudget(256); + using var channel = new TestChannel(budget); + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [1], 1); + switch (action) + { + case "dispose": channel.Dispose(); break; + case "close": channel.CloseChannel(); break; + case "fault": channel.Fault(); break; + case "disconnect": channel.OnReceiveError(null, new ServiceResult(StatusCodes.BadConnectionClosed)); break; + } + Assert.That(budget.ReservedBytes, Is.Zero); + Assert.That(channel.Pool.Outstanding, Is.Zero); + channel.Dispose(); + Assert.That(channel.Pool.Outstanding, Is.Zero); + } + + [Test] + public void FaultCleanupDoesNotPermanentlyDisableReassembly() + { + var budget = new ChunkReassemblyBudget(256); + using var channel = new TestChannel(budget); + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [], 1); + channel.Fault(); + channel.Reopen(); + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [], 2); + Assert.That(budget.ReservedBytes, Is.EqualTo(256)); + Assert.That(channel.Pool.Outstanding, Is.EqualTo(1)); + } + + [Test] + public void IntermediateCloseChunkIsReturnedExactlyOnce() + { + var budget = new ChunkReassemblyBudget(256); + using var channel = new TestChannel(budget); + channel.Receive(TcpMessageType.Close | TcpMessageType.Intermediate, [], 1); + Assert.That(channel.Closed, Is.True); + Assert.That(channel.Pool.Outstanding, Is.Zero); + Assert.That(budget.ReservedBytes, Is.Zero); + } + + [Test] + public void DiscoveryRejectionDoesNotLeakOrReadReturnedBuffers() + { + var budget = new ChunkReassemblyBudget(256); + using var channel = new TestChannel(budget); + channel.SetDiscoveryOnly(); + channel.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, channel.EncodeRequest(), 1); + Assert.That(channel.Requests, Is.Empty); + Assert.That(channel.Closed, Is.True); + Assert.That(channel.Pool.Outstanding, Is.Zero); + Assert.That(budget.ReservedBytes, Is.Zero); + } + + [TestCase(TcpMessageType.Intermediate)] + [TestCase(TcpMessageType.Final)] + public void MalformedFirstDiscoveryChunkReleasesItsBuffer(uint chunkType) + { + var budget = new ChunkReassemblyBudget(256); + using var channel = new TestChannel(budget); + channel.SetDiscoveryOnly(); + channel.Receive(TcpMessageType.Message | chunkType, [], 1); + Assert.That(channel.Requests, Is.Empty); + Assert.That(channel.Pool.Outstanding, Is.Zero); + Assert.That(budget.ReservedBytes, Is.Zero); + Assert.That(channel.SentMessages, Is.EqualTo(1), "The discovery fault response must still be sent."); + Assert.That(channel.DecodeLastResponse(), Is.TypeOf()); + Assert.That(channel.Closed, Is.False); + } + + [Test] + public void SavingWhileDisposingReturnsEveryBufferOnce() + { + for (int i = 0; i < 50; i++) + { + var budget = new ChunkReassemblyBudget(1024); + using var channel = new TestChannel(budget); + byte[] buffer = channel.Rent(100); + Parallel.Invoke(() => channel.Save(buffer), channel.Dispose); + Assert.That(channel.Pool.Outstanding, Is.Zero); + Assert.That(budget.ReservedBytes, Is.Zero); + } + } + + [Test] + public void ClientContextChunksDoNotConsumeServerBudget() + { + var budget = new ChunkReassemblyBudget(1); + using var channel = new TestChannel(budget); + channel.Save(channel.Rent(100), serverContext: false); + Assert.That(budget.ReservedBytes, Is.Zero); + Assert.That(channel.Pool.Outstanding, Is.EqualTo(1)); + } + + [Test] + public void StandaloneListenerCreatesDefaultBudgetAndTwoListenersShareInjectedBudget() + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + var budget = new ChunkReassemblyBudget(512); + using var first = new TcpTransportListener(telemetry); + using var second = new TcpTransportListener(telemetry); + using var standalone = new TcpTransportListener(telemetry); + Func membership = _ => false; + OpenListener(first, budget, telemetry, membership); + OpenListener(second, budget, telemetry, membership); + OpenListener(standalone, null, telemetry); + var firstQuotas = (ChannelQuotas)typeof(TcpTransportListener) + .GetField("m_quotas", BindingFlags.Instance | BindingFlags.NonPublic).GetValue(first); + var secondQuotas = (ChannelQuotas)typeof(TcpTransportListener) + .GetField("m_quotas", BindingFlags.Instance | BindingFlags.NonPublic).GetValue(second); + var standaloneQuotas = (ChannelQuotas)typeof(TcpTransportListener) + .GetField("m_quotas", BindingFlags.Instance | BindingFlags.NonPublic).GetValue(standalone); + Assert.That(firstQuotas.HasActivatedSession, Is.SameAs(membership)); + Assert.That(secondQuotas.HasActivatedSession, Is.SameAs(membership)); + Assert.That(standaloneQuotas.HasActivatedSession, Is.Null); + Assert.That(firstQuotas.ChunkReassemblyBudget, Is.SameAs(budget)); + Assert.That(secondQuotas.ChunkReassemblyBudget, Is.SameAs(budget)); + Assert.That(standaloneQuotas.ChunkReassemblyBudget.MaxBytes, Is.EqualTo(64L * 1024 * 1024)); + using var one = new TestChannel(budget, firstQuotas); + using var two = new TestChannel(budget, secondQuotas); + one.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [], 1); + two.Receive(TcpMessageType.Message | TcpMessageType.Intermediate, [], 1); + Assert.That(two.Closed, Is.True); + Assert.That(one.Closed, Is.False); + Assert.That(budget.ReservedBytes, Is.EqualTo(256)); + } + + [Test] + public async Task ServerSharesOneBudgetAcrossEndpointsAndResetsItAfterStopAsync() + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + using var server = new ServerBase(); + typeof(ServerBase).GetProperty("Configuration", BindingFlags.Instance | BindingFlags.NonPublic) + .SetValue(server, new ApplicationConfiguration(telemetry) { ServerConfiguration = new ServerConfiguration() }); + typeof(ServerBase).GetProperty("MessageContext") + .SetValue(server, new ServiceMessageContext(telemetry)); + Func membership = id => id == "activated"; + server.HasActivatedSession = membership; + var captured = new List(); + var capturedMembership = new List>(); + var listener = new Mock(); + listener.Setup(l => l.Open(It.IsAny(), It.IsAny(), It.IsAny())) + .Callback((uri, settings, callback) => + { + captured.Add(settings.ChunkReassemblyBudget); + capturedMembership.Add(settings.HasActivatedSession); + }); + var endpoint = new Uri("opc.tcp://localhost:12345"); + server.CreateServiceHostEndpoint(endpoint, [], EndpointConfiguration.Create(), listener.Object, null); + server.CreateServiceHostEndpoint(endpoint, [], EndpointConfiguration.Create(), listener.Object, null); + Assert.That(captured[0], Is.Not.Null); + Assert.That(captured[1], Is.SameAs(captured[0])); + Assert.That(capturedMembership.All(query => ReferenceEquals(query, membership)), Is.True); + await server.StopAsync().ConfigureAwait(false); + typeof(ServerBase).GetProperty("MessageContext") + .SetValue(server, new ServiceMessageContext(telemetry)); + server.CreateServiceHostEndpoint(endpoint, [], EndpointConfiguration.Create(), listener.Object, null); + Assert.That(captured[2], Is.Not.SameAs(captured[0])); + } + + [Test] + public async Task RealSocketsCannotAccumulateUnfinishedOpenMessagesAcrossListenersAsync() + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + var context = new ServiceMessageContext(telemetry); + var budget = new ChunkReassemblyBudget(32768); + using var first = new TcpTransportListener(telemetry); + using var second = new TcpTransportListener(telemetry); + OpenListener(first, budget, telemetry); + OpenListener(second, budget, telemetry); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(15)); + using var one = new TcpClient(); + using var two = new TcpClient(); + await one.ConnectAsync(IPAddress.Loopback, first.EndpointUrl.Port).ConfigureAwait(false); + await two.ConnectAsync(IPAddress.Loopback, second.EndpointUrl.Port).ConfigureAwait(false); + foreach ((TcpClient client, Uri endpoint) in new[] { (one, first.EndpointUrl), (two, second.EndpointUrl) }) + { + byte[] hello = EncodeFrame(context, encoder => + { + encoder.WriteUInt32(null, TcpMessageType.Hello); + encoder.WriteUInt32(null, 0); + encoder.WriteUInt32(null, 0); + encoder.WriteUInt32(null, 8192); + encoder.WriteUInt32(null, 8192); + encoder.WriteUInt32(null, 0); + encoder.WriteUInt32(null, 0); + encoder.WriteString(null, endpoint.ToString()); + }); + await WriteBytesAsync(client.GetStream(), hello, timeout.Token).ConfigureAwait(false); + byte[] acknowledge = await ReadFrameAsync(client.GetStream(), timeout.Token).ConfigureAwait(false); + Assert.That(BitConverter.ToUInt32(acknowledge, 0), Is.EqualTo(TcpMessageType.Acknowledge)); + } + byte[] partialOpen = EncodeFrame(context, encoder => + { + encoder.WriteUInt32(null, TcpMessageType.Open | TcpMessageType.Intermediate); + encoder.WriteUInt32(null, 0); + encoder.WriteUInt32(null, 0); + encoder.WriteString(null, SecurityPolicies.None); + encoder.WriteByteString(null, (byte[])null); + encoder.WriteByteString(null, (byte[])null); + encoder.WriteUInt32(null, 1); + encoder.WriteUInt32(null, 1); + }); + await WriteBytesAsync(one.GetStream(), partialOpen, timeout.Token).ConfigureAwait(false); + while (budget.ReservedBytes == 0) + { + await Task.Delay(10, timeout.Token).ConfigureAwait(false); + } + Assert.That(budget.ReservedBytes, Is.EqualTo(budget.MaxBytes / 2)); + await WriteBytesAsync(two.GetStream(), partialOpen, timeout.Token).ConfigureAwait(false); + byte[] error = await ReadFrameAsync(two.GetStream(), timeout.Token).ConfigureAwait(false); + Assert.That(BitConverter.ToUInt32(error, 0), Is.EqualTo(TcpMessageType.Error)); + Assert.That(BitConverter.ToUInt32(error, 8), Is.EqualTo(StatusCodes.BadTcpNotEnoughResources)); + Assert.That(budget.ReservedBytes, Is.EqualTo(budget.MaxBytes / 2)); + one.Close(); + while (budget.ReservedBytes != 0) + { + await Task.Delay(10, timeout.Token).ConfigureAwait(false); + } + } + + private static byte[] EncodeFrame(IServiceMessageContext context, Action write) + { + using var encoder = new BinaryEncoder(context); + write(encoder); + byte[] bytes = encoder.CloseAndReturnBuffer(); + BitConverter.GetBytes(bytes.Length).CopyTo(bytes, 4); + return bytes; + } + + private static Task WriteBytesAsync(NetworkStream stream, byte[] bytes, CancellationToken ct) + { +#if NET6_0_OR_GREATER + return stream.WriteAsync(bytes.AsMemory(), ct).AsTask(); +#else + return stream.WriteAsync(bytes, 0, bytes.Length, ct); +#endif + } + + private static async Task ReadFrameAsync(NetworkStream stream, CancellationToken ct) + { + byte[] header = new byte[8]; + await ReadBytesAsync(stream, header, 0, header.Length, ct).ConfigureAwait(false); + int size = BitConverter.ToInt32(header, 4); + Assert.That(size, Is.InRange(8, 65536)); + byte[] frame = new byte[size]; + header.CopyTo(frame, 0); + await ReadBytesAsync(stream, frame, 8, size - 8, ct).ConfigureAwait(false); + return frame; + } + + private static async Task ReadBytesAsync(NetworkStream stream, byte[] bytes, int offset, int length, CancellationToken ct) + { + while (length > 0) + { +#if NET6_0_OR_GREATER + int read = await stream.ReadAsync(bytes.AsMemory(offset, length), ct).ConfigureAwait(false); +#else + int read = await stream.ReadAsync(bytes, offset, length, ct).ConfigureAwait(false); +#endif + Assert.That(read, Is.GreaterThan(0), "Socket closed before sending the expected frame."); + offset += read; + length -= read; + } + } + + private static void OpenListener( + TcpTransportListener listener, + ChunkReassemblyBudget budget, + ITelemetryContext telemetry, + Func membership = null) + { + var context = new ServiceMessageContext(telemetry); + int port; + using (var probe = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp)) + { + probe.Bind(new IPEndPoint(IPAddress.Loopback, 0)); + port = ((IPEndPoint)probe.LocalEndPoint).Port; + } + var endpoint = new Uri($"opc.tcp://127.0.0.1:{port}"); + listener.Open(endpoint, new TransportListenerSettings + { + Configuration = EndpointConfiguration.Create(), + ServerCertificateTypesProvider = new CertificateTypesProvider( + new ApplicationConfiguration(telemetry) + { + SecurityConfiguration = new SecurityConfiguration + { + ApplicationCertificates = [new CertificateIdentifier + { + CertificateType = ObjectTypeIds.RsaSha256ApplicationCertificateType + }] + } + }, telemetry), + Descriptions = [new EndpointDescription(endpoint.ToString()) + { + SecurityMode = MessageSecurityMode.None, + SecurityPolicyUri = SecurityPolicies.None + }], + NamespaceUris = context.NamespaceUris, + Factory = context.Factory, + ChunkReassemblyBudget = budget, + HasActivatedSession = membership + }, null); + } + + private sealed class TrackingPool : ArrayPool + { + public int Outstanding + { + get { lock (m_lock) { return m_buffers.Count; } } + } + + public override byte[] Rent(int minimumLength) + { + var buffer = new byte[Math.Max(256, minimumLength)]; + lock (m_lock) { m_buffers.Add(buffer); } + return buffer; + } + + public override void Return(byte[] array, bool clearArray = false) + { + lock (m_lock) + { + Assert.That(m_buffers.Remove(array), Is.True, "Buffer returned twice or not rented here."); + // Expose accidental reads after return deterministically. + Array.Clear(array, 0, array.Length); + } + } + + private readonly object m_lock = new(); + private readonly HashSet m_buffers = []; + } + + private sealed class TestChannel : TcpServerChannel + { + public TestChannel(ChunkReassemblyBudget budget, ChannelQuotas quotas = null) + : this(budget, quotas, NUnitTelemetryContext.Create()) + { + } + + private TestChannel(ChunkReassemblyBudget budget, ChannelQuotas quotas, ITelemetryContext telemetry) + : base("test", new Mock().Object, + new BufferManager("test", 65535, telemetry), + quotas ?? new ChannelQuotas(new ServiceMessageContext(telemetry)) { ChunkReassemblyBudget = budget }, + null, [], telemetry) + { + Pool = new TrackingPool(); + typeof(BufferManager).GetField("m_arrayPool", BindingFlags.Instance | BindingFlags.NonPublic) + .SetValue(BufferManager, Pool); + ChannelId = 1; + UsedBySession = true; + var token = CreateToken(); + token.TokenId = 1; + ActivateToken(token); + State = TcpChannelState.Open; + SetRequestReceivedCallback((channel, id, request) => Requests.Add(request)); + var socket = new Mock(); + socket.Setup(s => s.MessageSocketEventArgs()).Returns(() => + { + if (FailReporting) { throw new InvalidOperationException("Simulated write failure"); } + var args = new Mock(); + args.SetupAllProperties(); + byte[] bytes = null; + int count = 0; + args.Setup(a => a.SetBuffer(It.IsAny(), It.IsAny(), It.IsAny())) + .Callback((b, offset, n) => + { + bytes = b; + count = n; + if (BitConverter.ToUInt32(b, offset) == TcpMessageType.Error) + { + LastError = BitConverter.ToUInt32(b, offset + 8); + } + }); + args.SetupGet(a => a.Buffer).Returns(() => bytes); + args.SetupGet(a => a.BytesTransferred).Returns(() => args.Object.BufferList?.TotalSize ?? count); + return args.Object; + }); + socket.Setup(s => s.Send(It.IsAny())) + .Callback(args => + { + SentMessages++; + // Copy before synchronous write completion returns and clears the buffers. + if (args.BufferList != null) + { + m_lastResponse = args.BufferList.SelectMany(buffer => buffer).ToArray(); + } + }) + .Returns(false); + Socket = socket.Object; + } + + public int SentMessages { get; private set; } + public IEncodeable DecodeLastResponse() + { + // TestChannel uses unsecured symmetric messages. + int headerSize = TcpMessageLimits.SymmetricHeaderSize + TcpMessageLimits.SequenceHeaderSize; + using var stream = new ArraySegmentStream(new BufferCollection( + new ArraySegment(m_lastResponse, headerSize, m_lastResponse.Length - headerSize))); + return BinaryDecoder.DecodeMessage(stream, null, Quotas.MessageContext); + } + + public TrackingPool Pool { get; } + public List Requests { get; } = []; + public bool Closed => State == TcpChannelState.Closed; + public bool FailReporting { get; set; } + public uint LastError { get; private set; } + public void SetSession(bool active) => UsedBySession = active; + public void SetDiscoveryOnly() => typeof(UaSCUaBinaryChannel) + .GetProperty("DiscoveryOnly", BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic) + .SetValue(this, true); + public void Reopen() => State = TcpChannelState.Open; + public void Fault() => ForceChannelFault(new ServiceResult(StatusCodes.BadConnectionClosed)); + public void CloseChannel() => ChannelClosed(); + public byte[] Rent(int size) => BufferManager.TakeBuffer(size, "test"); + public void Save(byte[] buffer, bool serverContext = true) => + SaveIntermediateChunk(1, new ArraySegment(buffer, 0, 1), serverContext); + public byte[] EncodeRequest() => BinaryEncoder.EncodeMessage(new ReadRequest + { + RequestHeader = new RequestHeader { RequestHandle = 1 }, + NodesToRead = [] + }, Quotas.MessageContext); + + public void Receive(uint type, byte[] body, uint requestId) + { + byte[] bytes = Rent(24 + body.Length); + using var encoder = new BinaryEncoder(bytes, 0, bytes.Length - 1, Quotas.MessageContext); + encoder.WriteUInt32(null, type); + encoder.WriteUInt32(null, (uint)(24 + body.Length)); + encoder.WriteUInt32(null, 1); + encoder.WriteUInt32(null, 1); + encoder.WriteUInt32(null, ++m_sequence); + encoder.WriteUInt32(null, requestId); + encoder.WriteRawBytes(body, 0, body.Length); + int length = encoder.Close(); + OnMessageReceived(Socket, new ArraySegment(bytes, 0, length)); + } + + private byte[] m_lastResponse; + private uint m_sequence; + } + } +} diff --git a/Tests/Opc.Ua.Server.Tests/SessionlessBudgetTests.cs b/Tests/Opc.Ua.Server.Tests/SessionlessBudgetTests.cs new file mode 100644 index 0000000000..63f4fc0455 --- /dev/null +++ b/Tests/Opc.Ua.Server.Tests/SessionlessBudgetTests.cs @@ -0,0 +1,322 @@ +/* ======================================================================== + * 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.Linq; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; +using Moq; +using NUnit.Framework; +using Opc.Ua.Tests; + +namespace Opc.Ua.Server.Tests +{ + [TestFixture] + [Category("ChunkReassembly")] + public class SessionlessBudgetTests + { + [Test] + public async Task LiveMembershipFollowsActivationTransferAndLastSessionClosureAsync() + { + var fixture = new ServerFixture { SecurityNone = true }; + await fixture.StartAsync().ConfigureAwait(false); + try + { + StandardServer server = fixture.Server; + var manager = (SessionManager)server.CurrentInstance.SessionManager; + Func membership = GetMembership(server); + var activatedCallbacks = new List(); + manager.SessionActivated += (session, _) => activatedCallbacks.Add(membership(session.SecureChannelId)); + (RequestHeader first, SecureChannelContext channel) = + await server.CreateAndActivateSessionAsync("original").ConfigureAwait(false); + (RequestHeader second, _) = await server.CreateAndActivateSessionAsync("original").ConfigureAwait(false); + Assert.That(membership("original"), Is.True); + Assert.That(activatedCallbacks, Is.Not.Empty); + Assert.That(activatedCallbacks.All(value => value), Is.True, "Membership must precede activation callbacks."); + + var transferred = new SecureChannelContext("transferred", channel.EndpointDescription, RequestEncoding.Binary); + var context = new OperationContext(first, transferred, RequestType.ActivateSession); + int eventsBefore = activatedCallbacks.Count; + (bool changed, _) = await manager.ActivateSessionAsync(context, first.AuthenticationToken, null, null, null, []) + .ConfigureAwait(false); + Assert.That(changed, Is.False, "Transfer with unchanged identity must still refresh membership."); + Assert.That(activatedCallbacks.Count, Is.EqualTo(eventsBefore), "Keep existing event semantics."); + Assert.That(membership("original"), Is.True, "Second session still belongs to the original channel."); + Assert.That(membership("transferred"), Is.True); + + bool? membershipDuringClose = null; + manager.SessionClosing += (session, _) => membershipDuringClose = membership(session.SecureChannelId); + manager.CloseSession(manager.GetSession(second.AuthenticationToken).Id); + Assert.That(membershipDuringClose, Is.False, "Removal must precede closing callbacks."); + Assert.That(membership("original"), Is.False); + Assert.That(membership("transferred"), Is.True); + manager.CloseSession(manager.GetSession(first.AuthenticationToken).Id); + Assert.That(membership("transferred"), Is.False); + Assert.That(manager.HasActivatedSession("original"), Is.False); + } + finally + { + await fixture.StopAsync().ConfigureAwait(false); + } + } + + [Test] + public async Task UnactivatedFailedAndExpiredSessionsDoNotKeepFullBudgetAccessAsync() + { + var fixture = new ServerFixture { SecurityNone = true }; + await fixture.StartAsync().ConfigureAwait(false); + try + { + StandardServer server = fixture.Server; + var manager = (SessionManager)server.CurrentInstance.SessionManager; + Func membership = GetMembership(server); + EndpointDescription endpoint = server.GetEndpoints().First(e => e.SecurityPolicyUri == SecurityPolicies.None); + var channel = new SecureChannelContext("unactivated", endpoint, RequestEncoding.Binary); + CreateSessionResponse created = await server.CreateSessionAsync(channel, new RequestHeader(), null, null, null, + "pending", null, null, 120000, 128 * 1024, CancellationToken.None).ConfigureAwait(false); + Assert.That(created.ResponseHeader.ServiceResult, Is.EqualTo(StatusCodes.Good)); + Assert.That(membership("unactivated"), Is.False); + var header = new RequestHeader { AuthenticationToken = created.AuthenticationToken }; + var wrongEndpoint = new EndpointDescription + { + SecurityMode = MessageSecurityMode.Sign, + SecurityPolicyUri = SecurityPolicies.Basic256Sha256 + }; + var wrongChannel = new SecureChannelContext("unactivated", wrongEndpoint, RequestEncoding.Binary); + Assert.ThrowsAsync(async () => + await manager.ActivateSessionAsync(new OperationContext(header, wrongChannel, RequestType.ActivateSession), + header.AuthenticationToken, null, null, null, []).ConfigureAwait(false)); + Assert.That(membership("unactivated"), Is.False); + + var context = new OperationContext(header, channel, RequestType.ActivateSession); + await manager.ActivateSessionAsync(context, header.AuthenticationToken, null, null, null, []).ConfigureAwait(false); + Assert.That(membership("unactivated"), Is.True); + var session = (Session)manager.GetSession(header.AuthenticationToken); + lock (session.DiagnosticsLock) + { + session.SessionDiagnostics.ClientLastContactTime = DateTime.UtcNow.AddDays(-1); + } + // Exercise the existing expired-session cleanup, without adding/waiting for a new timer. + Assert.ThrowsAsync(async () => + await manager.ActivateSessionAsync(context, header.AuthenticationToken, null, null, null, []).ConfigureAwait(false)); + Assert.That(manager.GetSession(header.AuthenticationToken), Is.Null); + Assert.That(membership("unactivated"), Is.False); + } + finally + { + await fixture.StopAsync().ConfigureAwait(false); + } + } + + [Test] + public async Task MembershipCallbackSurvivesStopAndRestartWithoutOldSessionsAsync() + { + var fixture = new ServerFixture { SecurityNone = true }; + await fixture.StartAsync().ConfigureAwait(false); + try + { + StandardServer server = fixture.Server; + Func membership = GetMembership(server); + await server.CreateAndActivateSessionAsync("old").ConfigureAwait(false); + var oldManager = (SessionManager)server.CurrentInstance.SessionManager; + Assert.That(membership("old"), Is.True); + await server.StopAsync().ConfigureAwait(false); + Assert.That(membership("old"), Is.False); + Assert.That(oldManager.HasActivatedSession("old"), Is.False); + await server.StartAsync(fixture.Config).ConfigureAwait(false); + Assert.That(membership("old"), Is.False); + await server.CreateAndActivateSessionAsync("new").ConfigureAwait(false); + Assert.That(membership("new"), Is.True); + Assert.That(GetMembership(server)("new"), Is.True); + } + finally + { + await fixture.StopAsync().ConfigureAwait(false); + } + } + + [TestCase(false)] + [TestCase(true)] + public void CustomManagersUseExistingSessionContractIncludingSubclasses(bool subclass) + { + using var server = new StandardServer(); + using SessionManager derived = subclass ? new DerivedSessionManager(CreateServer(), Configuration()) : null; + var session = new Mock(); + bool activated = false; + string channel = "first"; + session.SetupGet(s => s.Activated).Returns(() => activated); + session.SetupGet(s => s.SecureChannelId).Returns(() => channel); + var sessions = new List { session.Object }; + var custom = new Mock(); + custom.Setup(m => m.GetSessions()).Returns(sessions); + if (subclass) + { + Registry(derived).TryAdd(new NodeId(1), session.Object); + } + var membership = (Func)Delegate.CreateDelegate( + typeof(Func), server, "ChannelHasActivatedSession"); + Assert.That(membership("first"), Is.False, "No manager is available before startup."); + typeof(StandardServer).GetField("m_reassemblySessionManager", BindingFlags.NonPublic | BindingFlags.Instance) + .SetValue(server, subclass ? derived : custom.Object); + Assert.That(membership("first"), Is.False); + activated = true; + Assert.That(membership("first"), Is.True); + channel = "second"; + Assert.That(membership("first"), Is.False); + Assert.That(membership("second"), Is.True); + if (subclass) + { + Registry(derived).Clear(); + } + else + { + sessions.Clear(); + } + Assert.That(membership("second"), Is.False); + } + + [TestCase(false)] + [TestCase(true)] + public async Task ActivationFinishingAfterRemovalCannotRestoreMembershipAsync(bool shutdown) + { + using var manager = new SessionManager(CreateServer(), Configuration()); + var session = new Mock(); + session.SetupGet(s => s.Id).Returns(new NodeId(1)); + session.SetupGet(s => s.SecureChannelId).Returns("channel"); + bool activated = false; + session.SetupGet(s => s.Activated).Returns(() => Volatile.Read(ref activated)); + using var entered = new ManualResetEventSlim(); + using var resume = new ManualResetEventSlim(); + session.Setup(s => s.Activate(It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns(() => + { + Volatile.Write(ref activated, true); + entered.Set(); + if (!resume.Wait(TimeSpan.FromSeconds(10))) { throw new TimeoutException(); } + return false; + }); + var token = new NodeId(2); + Registry(manager).TryAdd(token, session.Object); + var context = new OperationContext(new RequestHeader(), new SecureChannelContext("channel", + new EndpointDescription { SecurityPolicyUri = SecurityPolicies.None }, RequestEncoding.Binary), RequestType.ActivateSession); + Task activation = Task.Run(async () => await manager.ActivateSessionAsync(context, token, null, null, null, []).ConfigureAwait(false)); + try + { + Assert.That(entered.Wait(TimeSpan.FromSeconds(10)), Is.True); + if (shutdown) { manager.Shutdown(); } + else { manager.CloseSession(session.Object.Id); } + Assert.That(manager.HasActivatedSession("channel"), Is.False); + } + finally + { + resume.Set(); + await activation.ConfigureAwait(false); + } + Assert.That(manager.HasActivatedSession("channel"), Is.False); + } + + [Test] + public async Task ConcurrentRefreshCannotOverwriteNewerMembershipAsync() + { + using var manager = new SessionManager(CreateServer(), Configuration()); + var registry = Registry(manager); + using var entered = new ManualResetEventSlim(); + using var resume = new ManualResetEventSlim(); + string channel = "old"; + int reads = 0; + var survivor = new Mock(); + survivor.SetupGet(s => s.Activated).Returns(true); + survivor.SetupGet(s => s.SecureChannelId).Returns(() => + { + string captured = Volatile.Read(ref channel); + if (Interlocked.Increment(ref reads) == 1) + { + entered.Set(); + if (!resume.Wait(TimeSpan.FromSeconds(10))) { throw new TimeoutException(); } + } + return captured; + }); + registry.TryAdd(new NodeId(1), survivor.Object); + for (uint id = 2; id <= 3; id++) + { + var removable = new Mock(); + removable.SetupGet(s => s.Id).Returns(new NodeId(id)); + registry.TryAdd(new NodeId(id), removable.Object); + } + Task first = Task.Run(() => manager.CloseSession(new NodeId(2))); + Task second = Task.CompletedTask; + try + { + Assert.That(entered.Wait(TimeSpan.FromSeconds(10)), Is.True); + Volatile.Write(ref channel, "new"); + second = Task.Run(() => manager.CloseSession(new NodeId(3))); + Assert.That(SpinWait.SpinUntil(() => !registry.ContainsKey(new NodeId(3)), TimeSpan.FromSeconds(10)), Is.True); + } + finally + { + resume.Set(); + await Task.WhenAll(first, second).ConfigureAwait(false); + } + Assert.That(manager.HasActivatedSession("old"), Is.False); + Assert.That(manager.HasActivatedSession("new"), Is.True); + manager.Dispose(); + Assert.That(manager.HasActivatedSession("new"), Is.False); + } + + private static Func GetMembership(StandardServer server) => + (Func)typeof(ServerBase).GetProperty("HasActivatedSession", BindingFlags.Instance | BindingFlags.NonPublic) + .GetValue(server); + + private static NodeIdDictionary Registry(SessionManager manager) => + (NodeIdDictionary)typeof(SessionManager).GetField("m_sessions", BindingFlags.Instance | BindingFlags.NonPublic) + .GetValue(manager); + + private static ApplicationConfiguration Configuration() => new() + { + ServerConfiguration = new ServerConfiguration() + }; + + private static IServerInternal CreateServer() + { + var server = new Mock(); + server.SetupGet(s => s.Telemetry).Returns(NUnitTelemetryContext.Create()); + server.SetupGet(s => s.DiagnosticsWriteLock).Returns(new object()); + server.SetupGet(s => s.ServerDiagnostics).Returns(new ServerDiagnosticsSummaryDataType { CurrentSessionCount = 100 }); + return server.Object; + } + + private sealed class DerivedSessionManager : SessionManager + { + public DerivedSessionManager(IServerInternal server, ApplicationConfiguration configuration) : base(server, configuration) { } + } + } +} From db39629d2a5b68cf5efd24f84a927ed55e04d9a9 Mon Sep 17 00:00:00 2001 From: Suciu Mircea Adrian Date: Tue, 29 Sep 2026 19:37:17 +0300 Subject: [PATCH 3/6] [CTT][master378] Fix ServerConfiguration URI values for CTT push-management checks (#4592) Populate ApplicationUri and ProductUri from application configuration, guarding absent optional nodes. Add push tests verifying both values match the advertised server identity. # Description Addresses the reported CTT complaint for Push Model for Global Certificate and TrustList Management, where these ServerConfiguration properties returned null: - ApplicationUri (ns=0;i=25706) - ProductUri (ns=0;i=25725) Initializes both properties from the application configuration, with guards for absent optional nodes. Validation: regression tests verify Good status and nonempty values matching the advertised server identity. All five variants pass; the full push fixture reports 116 passed, 9 expected skips, and no failures on .NET 10. CTT itself has not been rerun. ## Checklist _Put an `x` in the boxes that apply. You can complete these step by step after opening the PR._ - [X ] I have signed the [CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf) and read the [CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md) doc. - [X] I have added tests that prove my fix is effective or that my feature works and increased code coverage. - [ ] I have added all necessary documentation. - [X] I have verified that my changes do not introduce (new) build or analyzer warnings. - [ ] I ran **all** tests locally using the **UA.slnx** solution against at least .net **framework** and .net **10**, and all passed. - [ ] I fixed **all** failing and flaky tests in the CI pipelines and **all** CodeQL warnings. - [ ] I have addressed **all** PR feedback received. --- .../Configuration/ConfigurationNodeManager.cs | 9 +++++ Tests/Opc.Ua.Gds.Tests/PushTest.cs | 40 +++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/Libraries/Opc.Ua.Server/Configuration/ConfigurationNodeManager.cs b/Libraries/Opc.Ua.Server/Configuration/ConfigurationNodeManager.cs index a94b58aad7..0494e853ad 100644 --- a/Libraries/Opc.Ua.Server/Configuration/ConfigurationNodeManager.cs +++ b/Libraries/Opc.Ua.Server/Configuration/ConfigurationNodeManager.cs @@ -255,6 +255,15 @@ public void CreateServerConfiguration( ApplicationConfiguration configuration) { // setup server configuration node + if (m_serverConfigurationNode.ApplicationUri != null) + { + m_serverConfigurationNode.ApplicationUri.Value = configuration.ApplicationUri; + } + if (m_serverConfigurationNode.ProductUri != null) + { + m_serverConfigurationNode.ProductUri.Value = configuration.ProductUri; + } + m_serverConfigurationNode.ServerCapabilities.Value = [ .. configuration.ServerConfiguration.ServerCapabilities diff --git a/Tests/Opc.Ua.Gds.Tests/PushTest.cs b/Tests/Opc.Ua.Gds.Tests/PushTest.cs index 4bd1bf8a78..78e088caa9 100644 --- a/Tests/Opc.Ua.Gds.Tests/PushTest.cs +++ b/Tests/Opc.Ua.Gds.Tests/PushTest.cs @@ -243,6 +243,46 @@ public async Task GetSupportedKeyFormatsAsync() Assert.IsNotNull(keyFormats); } + [Test] + [Order(101)] + public async Task ReadServerConfigurationIdentityAsync() + { + await ConnectPushClientAsync(true).ConfigureAwait(false); + ISession session = m_pushClient.PushClient.Session; + ReadValueIdCollection nodesToRead = + [ + new ReadValueId + { + NodeId = OpcUa.VariableIds.ServerConfiguration_ApplicationUri, + AttributeId = Attributes.Value + }, + new ReadValueId + { + NodeId = OpcUa.VariableIds.ServerConfiguration_ProductUri, + AttributeId = Attributes.Value + } + ]; + + ReadResponse response = await session.ReadAsync( + null, + 0, + TimestampsToReturn.Neither, + nodesToRead, + CancellationToken.None).ConfigureAwait(false); + + Assert.AreEqual(nodesToRead.Count, response.Results.Count); + ApplicationDescription description = session.ConfiguredEndpoint.Description.Server; + string[] expectedUris = [description.ApplicationUri, description.ProductUri]; + for (int i = 0; i < nodesToRead.Count; i++) + { + DataValue result = response.Results[i]; + Assert.AreEqual(StatusCodes.Good, result.StatusCode.Code, $"{nodesToRead[i].NodeId}"); + Assert.IsInstanceOf(result.Value, $"{nodesToRead[i].NodeId}"); + Assert.IsNotEmpty((string)result.Value, $"{nodesToRead[i].NodeId}"); + Assert.AreEqual(expectedUris[i], result.Value, $"{nodesToRead[i].NodeId}"); + } + } + [Test] [Order(200)] public async Task ReadTrustListAsync() From 594eb7a2452a619702cba1f5d5b8b4e7c9237fc6 Mon Sep 17 00:00:00 2001 From: romanett Date: Thu, 1 Oct 2026 15:01:22 +0200 Subject: [PATCH 4/6] [master378] Bound nested array preallocation, XML depth and inbound PubSub metadata (#4603) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Minimal backports to the 1.5.378 line of three findings from the codec audit behind #4599 / #4602 (B1-1, X1-1, P1-1). All three can be reached from untrusted input. The changes are kept small, and the performance of legitimate messages is unchanged. ## Changes **B1-1: nested array preallocation (`BinaryDecoder`)** Variant, DataValue, ExtensionObject and encodeable arrays were allocated at the full length prefix before any element was read. Arrays nested in the first element of their parent all see the same remaining message, so a chain of them multiplied the allocation: about 1 KB of input in `RequestHeader.AdditionalHeader` preallocated about 200 MB before authentication. - These arrays now use the #4602 budget. An array is allocated at full length only while all such arrays still being read take at most 8 bytes per remaining message byte. Otherwise it starts at 16 KB and grows as elements are read. - The reservation is released in a `finally`. - Leaf arrays (numbers, strings, NodeIds, …) are unchanged and stay bounded by MaxArrayLength. **X1-1: XML element depth (`XmlDecoder`, `BinaryDecoder`, `JsonDecoder`)** The depth of XmlElement values and XML ExtensionObject bodies was unbounded, and the `InnerXml` setter and `CloneNode(true)` recurse once per level. A known-type XML body of about 200 KB inside a binary message terminated the process with a stack overflow. - Before any network XML is loaded into a DOM, it is now checked with a streaming reader. XML nested deeper than `MaxEncodingNestingLevels` (default 200, the same limit master uses) is rejected with `BadEncodingLimitsExceeded`. - Master's node-by-node copier is not ported. **P1-1: inbound DataSetMetaData (`UaPubSubConnection`)** Received metadata is not authenticated, yet it replaced the metadata of every reader with a matching DataSetWriterId, whichever publisher sent it. - It now only updates readers whose PublisherId filter accepts the sender, using the same rule as data messages. A reader without a PublisherId still accepts any publisher. - Master's registry cap is not needed, because 378 has no registry. ## Verification (net10.0 unless noted) | Check | Result | |---|---| | Nested Variant-array chain, 995 bytes | 199.2 MB → 3.3 MB allocated | | Known-type XML body 30 000 levels deep inside binary | process crash → `BadEncodingLimitsExceeded` | | `Opc.Ua.Types.Tests` (incl. new `DecoderNestingLimitTests`) | 327 / 327; new tests also 9 / 9 on net48 | | `Opc.Ua.Encoders.Fuzz.Tests` (crash/timeout/slow corpus) | 150 / 150 | | `Opc.Ua.Core.Tests` (Encoders + Types) | 6 327 passed, 0 failed | | `Opc.Ua.PubSub.Tests` (without broker-dependent MQTT/UDP) + new `MetaDataPublisherFilterTests` | 617 + 4 passed, 13 skipped | | Types / Core / PubSub build, all TFMs | 0 warnings | **Performance vs. db39629d2:** - Test cases: decoding `Variant[100000]`, `Variant(Variant[])`, `DataValue[10000]`, `ReadValueId[20000]` and `ExtensionObject[20000]`. - Allocations are byte-identical to the baseline. - Median times are within run-to-run noise over two interleaved rounds. - The only new cost is one extra streaming parse for XmlElement values. ## Notes for review - If a deployment raises `MaxEncodingNestingLevels` well above the default, the allowed XML depth rises with it. Master behaves the same way. - No public API changes. The XML depth helper is internal to `Opc.Ua.Types`, and `JsonDecoder` in `Opc.Ua.Core` has a private copy. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: romanett <7413710+romanett@users.noreply.github.com> --- Libraries/Opc.Ua.PubSub/UaPubSubConnection.cs | 35 ++- Stack/Opc.Ua.Core/Opc.Ua.Core.csproj | 3 + .../Opc.Ua.Core/Types/Encoders/JsonDecoder.cs | 2 + Stack/Opc.Ua.Types/Encoders/BinaryDecoder.cs | 235 +++++++++++----- Stack/Opc.Ua.Types/Encoders/XmlDecoder.cs | 6 +- .../Opc.Ua.Types/Encoders/XmlElementDepth.cs | 75 ++++++ .../Encoders/JsonDecoderXmlDepthTests.cs | 89 ++++++ .../Transport/MetaDataPublisherFilterTests.cs | 176 ++++++++++++ .../Encoders/DecoderNestingLimitTests.cs | 254 ++++++++++++++++++ 9 files changed, 807 insertions(+), 68 deletions(-) create mode 100644 Stack/Opc.Ua.Types/Encoders/XmlElementDepth.cs create mode 100644 Tests/Opc.Ua.Core.Tests/Types/Encoders/JsonDecoderXmlDepthTests.cs create mode 100644 Tests/Opc.Ua.PubSub.Tests/Transport/MetaDataPublisherFilterTests.cs create mode 100644 Tests/Opc.Ua.Types.Tests/Encoders/DecoderNestingLimitTests.cs diff --git a/Libraries/Opc.Ua.PubSub/UaPubSubConnection.cs b/Libraries/Opc.Ua.PubSub/UaPubSubConnection.cs index ed88b28472..bd3df74a6d 100644 --- a/Libraries/Opc.Ua.PubSub/UaPubSubConnection.cs +++ b/Libraries/Opc.Ua.PubSub/UaPubSubConnection.cs @@ -291,9 +291,12 @@ protected void ProcessDecodedNetworkMessage(UaNetworkMessage networkMessage, str lock (Lock) { - // check if reader's MetaData shall be updated + // check if reader's MetaData shall be updated. The metadata message is + // not authenticated: only accept it from the publisher the reader + // receives data from (6.2.8.1 PublisherId filter, null accepts any). if (reader.DataSetWriterId != 0 && reader.DataSetWriterId == networkMessage.DataSetWriterId && + MatchesPublisherId(reader, networkMessage) && ( reader.DataSetMetaData == null || !Utils.IsEqual( @@ -416,6 +419,36 @@ protected void ProcessDecodedNetworkMessage(UaNetworkMessage networkMessage, str } } + /// + /// Checks the PublisherId of a received network message against the PublisherId + /// filter of a reader, the same way the data messages are filtered when decoded. + /// A reader without PublisherId accepts any publisher. + /// + private static bool MatchesPublisherId( + DataSetReaderDataType reader, + UaNetworkMessage networkMessage) + { + object readerPublisherId = reader.PublisherId.Value; + if (readerPublisherId == null) + { + return true; + } + + switch (networkMessage) + { + case Encoding.JsonNetworkMessage jsonNetworkMessage: + return jsonNetworkMessage.PublisherId != null && + Utils.IsEqual( + jsonNetworkMessage.PublisherId, + readerPublisherId.ToString()); + case Encoding.UadpNetworkMessage uadpNetworkMessage: + return uadpNetworkMessage.PublisherId != null && + Utils.IsEqual(uadpNetworkMessage.PublisherId, readerPublisherId); + default: + return false; + } + } + /// /// Get all dataset readers defined for this UaSubscriber component /// diff --git a/Stack/Opc.Ua.Core/Opc.Ua.Core.csproj b/Stack/Opc.Ua.Core/Opc.Ua.Core.csproj index 7219f547b0..dd7b97e367 100644 --- a/Stack/Opc.Ua.Core/Opc.Ua.Core.csproj +++ b/Stack/Opc.Ua.Core/Opc.Ua.Core.csproj @@ -38,6 +38,9 @@ + + + diff --git a/Stack/Opc.Ua.Core/Types/Encoders/JsonDecoder.cs b/Stack/Opc.Ua.Core/Types/Encoders/JsonDecoder.cs index aa67fad5e1..ecf312d14e 100644 --- a/Stack/Opc.Ua.Core/Types/Encoders/JsonDecoder.cs +++ b/Stack/Opc.Ua.Core/Types/Encoders/JsonDecoder.cs @@ -776,6 +776,8 @@ public XmlElement ReadXmlElement(string fieldName) try { + XmlElementDepth.Check(value, Context, ConformanceLevel.Document); + var document = new XmlDocument(); using (var reader = XmlReader.Create( diff --git a/Stack/Opc.Ua.Types/Encoders/BinaryDecoder.cs b/Stack/Opc.Ua.Types/Encoders/BinaryDecoder.cs index e9f75c4ad3..2d315f4d3f 100644 --- a/Stack/Opc.Ua.Types/Encoders/BinaryDecoder.cs +++ b/Stack/Opc.Ua.Types/Encoders/BinaryDecoder.cs @@ -29,7 +29,6 @@ using System; using System.Collections.Generic; -using System.Globalization; using System.IO; using System.Runtime.CompilerServices; using System.Text; @@ -532,6 +531,7 @@ public XmlElement ReadXmlElement(string fieldName) utf8StringLength--; } string xmlString = Encoding.UTF8.GetString(bytes, 0, utf8StringLength); + XmlElementDepth.Check(xmlString, Context, ConformanceLevel.Document); using var stream = new StringReader(xmlString); using var reader = XmlReader.Create(stream, CoreUtils.DefaultXmlReaderSettings()); document.Load(reader); @@ -1223,14 +1223,22 @@ public VariantCollection ReadVariantArray(string fieldName) return null; } - var values = new VariantCollection(length); + int capacity = ReserveArrayCapacity(length, s_variantElementSize, out long reserved); + try + { + var values = new VariantCollection(capacity); - for (int ii = 0; ii < length; ii++) + for (int ii = 0; ii < length; ii++) + { + values.Add(ReadVariant(null)); + } + + return values; + } + finally { - values.Add(ReadVariant(null)); + m_preallocatedBytes -= reserved; } - - return values; } /// @@ -1243,14 +1251,22 @@ public DataValueCollection ReadDataValueArray(string fieldName) return null; } - var values = new DataValueCollection(length); + int capacity = ReserveArrayCapacity(length, IntPtr.Size, out long reserved); + try + { + var values = new DataValueCollection(capacity); - for (int ii = 0; ii < length; ii++) + for (int ii = 0; ii < length; ii++) + { + values.Add(ReadDataValue(null)); + } + + return values; + } + finally { - values.Add(ReadDataValue(null)); + m_preallocatedBytes -= reserved; } - - return values; } /// @@ -1263,14 +1279,22 @@ public ExtensionObjectCollection ReadExtensionObjectArray(string fieldName) return null; } - var values = new ExtensionObjectCollection(length); + int capacity = ReserveArrayCapacity(length, IntPtr.Size, out long reserved); + try + { + var values = new ExtensionObjectCollection(capacity); + + for (int ii = 0; ii < length; ii++) + { + values.Add(ReadExtensionObject(null)); + } - for (int ii = 0; ii < length; ii++) + return values; + } + finally { - values.Add(ReadExtensionObject(null)); + m_preallocatedBytes -= reserved; } - - return values; } /// @@ -1286,14 +1310,7 @@ public Array ReadEncodeableArray( return null; } - var values = Array.CreateInstance(systemType, length); - - for (int ii = 0; ii < length; ii++) - { - values.SetValue(ReadEncodeable(null, systemType, encodeableTypeId), ii); - } - - return values; + return ReadEncodeableElements(length, systemType, encodeableTypeId); } /// @@ -1426,18 +1443,7 @@ public Array ReadArray( Array elements = null; if (DetermineIEncodeableSystemType(ref systemType, encodeableTypeId)) { - elements = Array.CreateInstance(systemType, length); - for (int i = 0; i < length; i++) - { - IEncodeable element = ReadEncodeable( - null, - systemType, - encodeableTypeId); - elements.SetValue(Convert.ChangeType( - element, - systemType, - CultureInfo.InvariantCulture), i); - } + elements = ReadEncodeableElements(length, systemType, encodeableTypeId); } elements ??= ReadArrayElements(length, builtInType); @@ -1849,41 +1855,23 @@ private Array ReadArrayElements(int length, BuiltInType builtInType) break; } case BuiltInType.ExtensionObject: - { - var values = new ExtensionObject[length]; - - for (int ii = 0; ii < values.Length; ii++) - { - values[ii] = ReadExtensionObject(); - } - - array = values; + array = ReadNestedElements( + length, + IntPtr.Size, + static decoder => decoder.ReadExtensionObject()); break; - } case BuiltInType.DataValue: - { - var values = new DataValue[length]; - - for (int ii = 0; ii < values.Length; ii++) - { - values[ii] = ReadDataValue(null); - } - - array = values; + array = ReadNestedElements( + length, + IntPtr.Size, + static decoder => decoder.ReadDataValue(null)); break; - } case BuiltInType.Variant: - { - var values = new Variant[length]; - - for (int ii = 0; ii < values.Length; ii++) - { - values[ii] = ReadVariant(null); - } - - array = values; + array = ReadNestedElements( + length, + s_variantElementSize, + static decoder => decoder.ReadVariant(null)); break; - } case BuiltInType.DiagnosticInfo: { var values = new DiagnosticInfo[length]; @@ -1938,6 +1926,103 @@ private int ReadArrayLength([CallerMemberName] string callerMemberName = "") return length; } + /// + /// Returns how many elements to allocate up front for an array of + /// elements that can contain nested arrays. + /// The length prefix of such an array is not checked against the + /// remaining message, and every level of a chain of arrays nested in + /// the first element of their parent sees the same remaining bytes, + /// so the preallocations of a chain would multiply. An array is + /// therefore allocated at its full length only while all arrays + /// allocated this way and still being read, this one included, take + /// at most bytes + /// per remaining message byte. Otherwise the caller starts with + /// and grows the array as the + /// elements are read. The caller subtracts + /// from after the last element. + /// + private int ReserveArrayCapacity(int length, int elementSize, out long reserved) + { + reserved = 0; + long bytes = (long)length * elementSize; + if (bytes <= kMaxPreallocatedArrayBytes) + { + return length; + } + + Stream stream = m_reader.BaseStream; + if (stream.CanSeek && + m_preallocatedBytes + bytes <= + kMaxPreallocatedBytesPerRemainingByte * Math.Max(0, stream.Length - stream.Position)) + { + reserved = bytes; + m_preallocatedBytes += bytes; + return length; + } + + return kMaxPreallocatedArrayBytes / elementSize; + } + + /// + /// Reads an array of values that can contain nested arrays. + /// + private T[] ReadNestedElements( + int length, + int elementSize, + Func readElement) + { + int capacity = ReserveArrayCapacity(length, elementSize, out long reserved); + try + { + var values = new T[capacity]; + for (int ii = 0; ii < length; ii++) + { + if (ii == values.Length) + { + Array.Resize(ref values, (int)Math.Min(length, 2L * values.Length)); + } + values[ii] = readElement(this); + } + return values; + } + finally + { + m_preallocatedBytes -= reserved; + } + } + + /// + /// Reads an array of encodeables of . + /// + private Array ReadEncodeableElements( + int length, + Type systemType, + ExpandedNodeId encodeableTypeId) + { + int capacity = ReserveArrayCapacity(length, IntPtr.Size, out long reserved); + try + { + Array values = Array.CreateInstance(systemType, capacity); + for (int ii = 0; ii < length; ii++) + { + if (ii == values.Length) + { + Array grown = Array.CreateInstance( + systemType, + (int)Math.Min(length, 2L * values.Length)); + Array.Copy(values, grown, values.Length); + values = grown; + } + values.SetValue(ReadEncodeable(null, systemType, encodeableTypeId), ii); + } + return values; + } + finally + { + m_preallocatedBytes -= reserved; + } + } + /// /// Reads the body of a node id. /// @@ -2684,11 +2769,29 @@ private static void ValidateStreamRequirements(Stream stream) } } + /// + /// Arrays of nested values up to this size are always allocated at + /// their full length. + /// + private const int kMaxPreallocatedArrayBytes = 16 * 1024; + + /// + /// The preallocation budget shared by the arrays of nested values + /// that are being read, per remaining message byte. + /// + private const int kMaxPreallocatedBytesPerRemainingByte = 8; + + /// + /// The size of a Variant array element (a value and a type reference). + /// + private static readonly int s_variantElementSize = 2 * IntPtr.Size; + private BinaryReader m_reader; private ushort[] m_namespaceMappings; private ushort[] m_serverMappings; private uint m_nestingLevel; private uint m_encodeablesRecovered; + private long m_preallocatedBytes; private readonly ILogger m_logger; } } diff --git a/Stack/Opc.Ua.Types/Encoders/XmlDecoder.cs b/Stack/Opc.Ua.Types/Encoders/XmlDecoder.cs index 928d60545d..14d937ea1f 100644 --- a/Stack/Opc.Ua.Types/Encoders/XmlDecoder.cs +++ b/Stack/Opc.Ua.Types/Encoders/XmlDecoder.cs @@ -591,6 +591,8 @@ public object ReadExtensionObjectBody(ExpandedNodeId typeId) ae.Message); } + XmlElementDepth.Check(xmlString, Context, ConformanceLevel.Document); + // check for empty body. var document = new XmlDocument(); @@ -1081,7 +1083,9 @@ public XmlElement ReadXmlElement(string fieldName) m_reader.MoveToContent(); } - value.InnerXml = m_reader.ReadInnerXml(); + string innerXml = m_reader.ReadInnerXml(); + XmlElementDepth.Check(innerXml, Context, ConformanceLevel.Fragment); + value.InnerXml = innerXml; EndField(fieldName); return value; diff --git a/Stack/Opc.Ua.Types/Encoders/XmlElementDepth.cs b/Stack/Opc.Ua.Types/Encoders/XmlElementDepth.cs new file mode 100644 index 0000000000..cb2730d248 --- /dev/null +++ b/Stack/Opc.Ua.Types/Encoders/XmlElementDepth.cs @@ -0,0 +1,75 @@ +/* ======================================================================== + * 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.IO; +using System.Xml; +using Opc.Ua.Types; + +namespace Opc.Ua +{ + /// + /// Checks XML element depth before loading untrusted XML into a DOM. + /// + internal static class XmlElementDepth + { + /// + /// Rejects XML whose elements are nested deeper than the encoding limit. + /// + internal static void Check( + string xml, + IServiceMessageContext context, + ConformanceLevel conformanceLevel) + { + if (string.IsNullOrEmpty(xml)) + { + return; + } + + int maxDepth = context.MaxEncodingNestingLevels > 0 + ? context.MaxEncodingNestingLevels + : DefaultEncodingLimits.MaxEncodingNestingLevels; + + XmlReaderSettings settings = CoreUtils.DefaultXmlReaderSettings(); + settings.ConformanceLevel = conformanceLevel; + + using var stream = new StringReader(xml); + using var reader = XmlReader.Create(stream, settings); + while (reader.Read()) + { + if (reader.NodeType == XmlNodeType.Element && reader.Depth > maxDepth) + { + throw ServiceResultException.Create( + StatusCodes.BadEncodingLimitsExceeded, + "XML element nesting exceeds the maximum depth of {0}.", + maxDepth); + } + } + } + } +} diff --git a/Tests/Opc.Ua.Core.Tests/Types/Encoders/JsonDecoderXmlDepthTests.cs b/Tests/Opc.Ua.Core.Tests/Types/Encoders/JsonDecoderXmlDepthTests.cs new file mode 100644 index 0000000000..e160aba6be --- /dev/null +++ b/Tests/Opc.Ua.Core.Tests/Types/Encoders/JsonDecoderXmlDepthTests.cs @@ -0,0 +1,89 @@ +/* ======================================================================== + * 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.Text; +using NUnit.Framework; +using Opc.Ua.Tests; + +namespace Opc.Ua.Core.Tests.Types.Encoders +{ + /// + /// Tests that the JSON decoder rejects deeply nested XmlElement values + /// before recursive DOM operations on them could overflow the stack. + /// + [TestFixture] + [Category("JsonEncoder")] + [SetCulture("en-us")] + [SetUICulture("en-us")] + [Parallelizable] + public class JsonDecoderXmlDepthTests + { + [Test] + [TestCase(50, false)] + [TestCase(300, true)] + [TestCase(30000, true)] + public void JsonDecoderRejectsDeeplyNestedXmlElement(int depth, bool rejected) + { + var context = new ServiceMessageContext(NUnitTelemetryContext.Create()); + var xml = new StringBuilder(""); + for (int ii = 0; ii < depth; ii++) + { + xml.Append(""); + } + for (int ii = 0; ii < depth; ii++) + { + xml.Append(""); + } + xml.Append(""); + + using var decoder = new JsonDecoder($"{{\"Value\":\"{xml}\"}}", context); + if (rejected) + { + ServiceResultException sre = Assert.Throws( + () => decoder.ReadXmlElement("Value")); + Assert.That(sre.StatusCode, Is.EqualTo(StatusCodes.BadEncodingLimitsExceeded)); + } + else + { + Assert.That(decoder.ReadXmlElement("Value").LocalName, Is.EqualTo("r")); + } + } + + [Test] + public void JsonDecoderRejectsMalformedXmlElement() + { + var context = new ServiceMessageContext(NUnitTelemetryContext.Create()); + using var decoder = new JsonDecoder("{\"Value\":\"\"}", context); + + ServiceResultException sre = Assert.Throws( + () => decoder.ReadXmlElement("Value")); + Assert.That(sre.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); + } + } +} diff --git a/Tests/Opc.Ua.PubSub.Tests/Transport/MetaDataPublisherFilterTests.cs b/Tests/Opc.Ua.PubSub.Tests/Transport/MetaDataPublisherFilterTests.cs new file mode 100644 index 0000000000..7292881614 --- /dev/null +++ b/Tests/Opc.Ua.PubSub.Tests/Transport/MetaDataPublisherFilterTests.cs @@ -0,0 +1,176 @@ +/* ======================================================================== + * 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.Reflection; +using NUnit.Framework; +using Opc.Ua.Tests; +using PubSubEncoding = Opc.Ua.PubSub.Encoding; + +namespace Opc.Ua.PubSub.Tests.Transport +{ + /// + /// Received DataSetMetaData is not authenticated, so it may only update + /// the readers that receive data from the publisher that sent it. + /// + [TestFixture(Description = "Tests the PublisherId filter for received metadata")] + public class MetaDataPublisherFilterTests + { + private const ushort kDataSetWriterId = 1; + + [Test] + [TestCase("Publisher1", "Publisher1", true)] + [TestCase("Publisher1", "Attacker", false)] + [TestCase("Publisher1", null, false)] + [TestCase(null, "Anyone", true)] + public void JsonMetaDataUpdatesOnlyReadersOfTheSendingPublisher( + string readerPublisherId, + string messagePublisherId, + bool expectUpdate) + { + DataSetMetaDataType metaData = CreateMetaData(); + var networkMessage = new PubSubEncoding.JsonNetworkMessage(null, metaData) + { + PublisherId = messagePublisherId, + DataSetWriterId = kDataSetWriterId + }; + + DataSetReaderDataType reader = ProcessMetaData( + readerPublisherId == null ? Variant.Null : new Variant(readerPublisherId), + Profiles.PubSubMqttJsonTransport, + networkMessage); + + AssertUpdated(reader, metaData, expectUpdate); + } + + [Test] + [TestCase((ushort)10, (ushort)10, true)] + [TestCase((ushort)10, (ushort)11, false)] + [TestCase((ushort)10, null, false)] + [TestCase(null, (ushort)11, true)] + public void UadpMetaDataUpdatesOnlyReadersOfTheSendingPublisher( + object readerPublisherId, + object messagePublisherId, + bool expectUpdate) + { + DataSetMetaDataType metaData = CreateMetaData(); + var networkMessage = new PubSubEncoding.UadpNetworkMessage(null, metaData) + { + PublisherId = messagePublisherId, + DataSetWriterId = kDataSetWriterId + }; + + DataSetReaderDataType reader = ProcessMetaData( + readerPublisherId == null ? Variant.Null : new Variant(readerPublisherId), + Profiles.PubSubMqttUadpTransport, + networkMessage); + + AssertUpdated(reader, metaData, expectUpdate); + } + + private static DataSetMetaDataType CreateMetaData() + { + return new DataSetMetaDataType + { + Name = "DataSet1", + ConfigurationVersion = new ConfigurationVersionDataType + { + MajorVersion = 1, + MinorVersion = 1 + } + }; + } + + private static void AssertUpdated( + DataSetReaderDataType reader, + DataSetMetaDataType metaData, + bool expectUpdate) + { + if (expectUpdate) + { + Assert.That(reader.DataSetMetaData, Is.SameAs(metaData)); + } + else + { + Assert.That(reader.DataSetMetaData, Is.Not.SameAs(metaData)); + } + } + + /// + /// Passes a received metadata message to a connection with one reader + /// and returns the reader. + /// + private static DataSetReaderDataType ProcessMetaData( + Variant readerPublisherId, + string transportProfileUri, + UaNetworkMessage networkMessage) + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + var reader = new DataSetReaderDataType + { + Name = "Reader1", + Enabled = true, + DataSetWriterId = kDataSetWriterId, + PublisherId = readerPublisherId + }; + var configuration = new PubSubConfigurationDataType + { + Enabled = true, + Connections = + [ + new PubSubConnectionDataType + { + Name = "Connection1", + Enabled = true, + PublisherId = new Variant("Subscriber1"), + TransportProfileUri = transportProfileUri, + Address = new ExtensionObject( + new NetworkAddressUrlDataType { Url = "mqtt://localhost:1883" }), + ReaderGroups = + [ + new ReaderGroupDataType + { + Name = "ReaderGroup1", + Enabled = true, + DataSetReaders = [reader] + } + ] + } + ] + }; + var application = UaPubSubApplication.Create(configuration, telemetry); + var connection = (UaPubSubConnection)application.PubSubConnections[0]; + + MethodInfo process = typeof(UaPubSubConnection).GetMethod( + "ProcessDecodedNetworkMessage", + BindingFlags.Instance | BindingFlags.NonPublic); + process.Invoke(connection, [networkMessage, "test"]); + return reader; + } + } +} diff --git a/Tests/Opc.Ua.Types.Tests/Encoders/DecoderNestingLimitTests.cs b/Tests/Opc.Ua.Types.Tests/Encoders/DecoderNestingLimitTests.cs new file mode 100644 index 0000000000..97fe0ffbcc --- /dev/null +++ b/Tests/Opc.Ua.Types.Tests/Encoders/DecoderNestingLimitTests.cs @@ -0,0 +1,254 @@ +/* ======================================================================== + * 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.IO; +using System.Text; +using System.Xml; +using NUnit.Framework; +using Opc.Ua.Tests; + +namespace Opc.Ua.Types.Tests.Encoders +{ + /// + /// Tests that nested arrays and nested XML read from untrusted input + /// stay within the memory and stack the input can justify. + /// + [TestFixture] + [Category("Encoders")] + [SetCulture("en-us")] + [SetUICulture("en-us")] + [Parallelizable] + public class DecoderNestingLimitTests + { + private const string kTypesNamespace = "http://opcfoundation.org/UA/2008/02/Types.xsd"; + + /// + /// A chain of Variant arrays whose length prefixes promise 65535 + /// elements each, nested in the first element of the parent, used to + /// preallocate about 1 MB per 5 input bytes and level. + /// + [Test] + public void NestedVariantArrayLengthPrefixesDoNotMultiplyAllocations() + { + var context = new ServiceMessageContext(NUnitTelemetryContext.Create()); + byte[] message = CreateNestedVariantArrayChain(199); + +#if NET + long before = GC.GetAllocatedBytesForCurrentThread(); +#endif + ServiceResultException sre = Assert.Throws(() => + { + using var decoder = new BinaryDecoder(message, context); + decoder.ReadVariant(null); + }); + Assert.That(sre.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); +#if NET + long allocated = GC.GetAllocatedBytesForCurrentThread() - before; + // 199 levels of 16 KB growing arrays, not 199 x 1 MB. + Assert.That(allocated, Is.LessThan(16 * 1024 * 1024)); +#endif + } + + /// + /// An array of null Variants takes one byte per element on the wire + /// and 16 bytes in memory, more than the preallocation budget, so it + /// is grown while it is read and still decodes completely. + /// + [Test] + public void VariantArrayLargerThanPreallocationBudgetDecodes() + { + var context = new ServiceMessageContext(NUnitTelemetryContext.Create()); + const int length = 65535; + using var stream = new MemoryStream(); + using (var writer = new BinaryWriter(stream, Encoding.UTF8, true)) + { + writer.Write((byte)(0x80 | (byte)BuiltInType.Variant)); + writer.Write(length); + writer.Write(new byte[length]); + } + + using var decoder = new BinaryDecoder(stream.ToArray(), context); + Variant variant = decoder.ReadVariant(null); + + var values = (Variant[])variant.Value; + Assert.That(values, Has.Length.EqualTo(length)); + Assert.That(values[length - 1].Value, Is.Null); + + // The same array as a VariantArray field, without the Variant encoding byte. + byte[] field = stream.ToArray(); + using var decoder2 = new BinaryDecoder(field, 1, field.Length - 1, context); + Assert.That(decoder2.ReadVariantArray(null), Has.Count.EqualTo(length)); + } + + /// + /// An encodeable array whose elements take no bytes on the wire is + /// grown while it is read and still decodes completely. + /// + [Test] + public void EncodeableArrayLargerThanPreallocationBudgetDecodes() + { + var context = new ServiceMessageContext(NUnitTelemetryContext.Create()); + const int length = 20000; + byte[] message = BitConverter.GetBytes(length); + + using var decoder = new BinaryDecoder(message, context); + Array values = decoder.ReadEncodeableArray(null, typeof(EmptyEncodeable)); + + Assert.That(values, Is.TypeOf()); + Assert.That(values, Has.Length.EqualTo(length)); + Assert.That(values.GetValue(length - 1), Is.Not.Null); + } + + /// + /// A deeply nested XmlElement in a binary message is rejected before + /// it is loaded; recursive DOM operations on it would overflow the stack. + /// + [Test] + [TestCase(50, false)] + [TestCase(300, true)] + [TestCase(30000, true)] + public void BinaryDecoderRejectsDeeplyNestedXmlElement(int depth, bool rejected) + { + var context = new ServiceMessageContext(NUnitTelemetryContext.Create()); + byte[] xml = Encoding.UTF8.GetBytes(CreateNestedXml(depth)); + using var stream = new MemoryStream(); + using (var writer = new BinaryWriter(stream, Encoding.UTF8, true)) + { + writer.Write(xml.Length); + writer.Write(xml); + } + + using var decoder = new BinaryDecoder(stream.ToArray(), context); + if (rejected) + { + ServiceResultException sre = Assert.Throws( + () => decoder.ReadXmlElement(null)); + Assert.That(sre.StatusCode, Is.EqualTo(StatusCodes.BadEncodingLimitsExceeded)); + } + else + { + Assert.That(decoder.ReadXmlElement(null).LocalName, Is.EqualTo("r")); + } + } + + /// + /// A deeply nested XmlElement in an XML document is rejected before + /// the InnerXml setter, which recurses once per level, parses it. + /// + [Test] + [TestCase(50, false)] + [TestCase(300, true)] + [TestCase(30000, true)] + public void XmlDecoderRejectsDeeplyNestedXmlElement(int depth, bool rejected) + { + var context = new ServiceMessageContext(NUnitTelemetryContext.Create()); + string document = + $"{CreateNestedXml(depth)}"; + + using var reader = XmlReader.Create(new StringReader(document)); + using var decoder = new XmlDecoder(null, reader, context); + if (rejected) + { + ServiceResultException sre = Assert.Throws( + () => decoder.ReadXmlElement("Value")); + Assert.That(sre.StatusCode, Is.EqualTo(StatusCodes.BadEncodingLimitsExceeded)); + } + else + { + Assert.That(decoder.ReadXmlElement("Value").LocalName, Is.EqualTo("r")); + } + } + + private static byte[] CreateNestedVariantArrayChain(int levels) + { + using var stream = new MemoryStream(); + using (var writer = new BinaryWriter(stream, Encoding.UTF8, true)) + { + for (int ii = 0; ii < levels; ii++) + { + writer.Write((byte)(0x80 | (byte)BuiltInType.Variant)); + writer.Write(65535); + } + } + return stream.ToArray(); + } + + private static string CreateNestedXml(int depth) + { + var builder = new StringBuilder(""); + for (int ii = 0; ii < depth; ii++) + { + builder.Append(""); + } + for (int ii = 0; ii < depth; ii++) + { + builder.Append(""); + } + return builder.Append("").ToString(); + } + + /// + /// An encodeable without fields. + /// + public sealed class EmptyEncodeable : IEncodeable + { + /// + public ExpandedNodeId TypeId => ExpandedNodeId.Null; + + /// + public ExpandedNodeId BinaryEncodingId => ExpandedNodeId.Null; + + /// + public ExpandedNodeId XmlEncodingId => ExpandedNodeId.Null; + + /// + public void Encode(IEncoder encoder) + { + } + + /// + public void Decode(IDecoder decoder) + { + } + + /// + public bool IsEqual(IEncodeable encodeable) + { + return encodeable is EmptyEncodeable; + } + + /// + public object Clone() + { + return new EmptyEncodeable(); + } + } + } +} From 3b3305355b914fb3472421640e1ffdaf997e22e9 Mon Sep 17 00:00:00 2001 From: Suciu Mircea Adrian Date: Fri, 2 Oct 2026 08:24:23 +0300 Subject: [PATCH 5/6] [master378] Fix queued data delivery after durable subscription transfer (#4622) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Force a fresh queue check after transfer so the first Publish returns saved data instead of an empty keep-alive. Reset the subscription’s expiration counter after transfer. Add tests for queued data delivery, the timing bug, and an empty keep-alive after all queued values have been delivered. # Description A durable subscription could return an empty keep-alive on the first Publish after transfer, even when values had accumulated while the client was disconnected. The server could reuse its previous keep-alive state before checking the queues. This change forces a fresh queue check after transfer and resets the subscription's expiration counter. Queued values are delivered first; normal keep-alives resume once the queues are drained. Tests cover the timing bug, data delivery after reconnecting with SendInitialValues=false, and a keep-alive after draining all queued writes, including across multiple Publish responses. ## Checklist _Put an `x` in the boxes that apply. You can complete these step by step after opening the PR._ - [x] I have signed the [CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf) and read the [CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md) doc. - [x] I have added tests that prove my fix is effective or that my feature works and increased code coverage. - [ ] I have added all necessary documentation. - [ ] I have verified that my changes do not introduce (new) build or analyzer warnings. - [ ] I ran **all** tests locally using the **UA.slnx** solution against at least .net **framework** and .net **10**, and all passed. - [ ] I fixed **all** failing and flaky tests in the CI pipelines and **all** CodeQL warnings. - [ ] I have addressed **all** PR feedback received. --- .../Subscription/Subscription.cs | 11 + .../DurableSubscriptionTest.cs | 224 ++++++++++++++++++ .../Opc.Ua.Server.Tests/SubscriptionTests.cs | 54 +++++ 3 files changed, 289 insertions(+) diff --git a/Libraries/Opc.Ua.Server/Subscription/Subscription.cs b/Libraries/Opc.Ua.Server/Subscription/Subscription.cs index e706f842a8..b8fe0c9b90 100644 --- a/Libraries/Opc.Ua.Server/Subscription/Subscription.cs +++ b/Libraries/Opc.Ua.Server/Subscription/Subscription.cs @@ -606,6 +606,17 @@ await m_server.NodeManager m_logger.LogTrace("Failed to transfer {Count} Monitored Items", badTransfers); } + lock (m_lock) + { + ResetLifetimeCount(); + + // An abandoned subscription can be waiting for a keep-alive without having + // checked its monitored item queues. Force the next timer check to inspect + // queued data before the new session's first Publish, even when initial + // values were not requested. + m_publishTimerExpiry = HiResClock.TickCount64 - 1; + } + lock (DiagnosticsWriteLock) { Diagnostics.SessionId = Session.Id; diff --git a/Tests/Opc.Ua.Client.Tests/DurableSubscriptionTest.cs b/Tests/Opc.Ua.Client.Tests/DurableSubscriptionTest.cs index f1a22121d3..714efa85aa 100644 --- a/Tests/Opc.Ua.Client.Tests/DurableSubscriptionTest.cs +++ b/Tests/Opc.Ua.Client.Tests/DurableSubscriptionTest.cs @@ -30,7 +30,9 @@ using System; using System.Collections.Generic; using System.Globalization; +using System.Linq; using System.Runtime.InteropServices; +using System.Threading; using System.Threading.Tasks; using NUnit.Framework; using Opc.Ua.Server.Tests; @@ -613,6 +615,228 @@ private async Task> ValidateDataValueAsync( return modifiedValues; } + /// + /// Queued samples must be sent on the first Publish after TransferSubscriptions(false). + /// Raw services prevent the client's automatic Publish pipeline from consuming the response. + /// + [Test] + public async Task TransferWithoutInitialValuesPublishesQueuedCurrentTimeAsync() + { + uint id = await CreateRawDurableSubscriptionAsync(0).ConfigureAwait(false); + try + { + CreateMonitoredItemsResponse items = await CreateRawMonitoredItemsAsync( + id, [VariableIds.Server_ServerStatus_CurrentTime], 1000, 1000).ConfigureAwait(false); + await Task.Delay(TimeSpan.FromMilliseconds(items.Results[0].RevisedSamplingInterval * 1.5)) + .ConfigureAwait(false); + List initial = await DrainRawAsync(id, + await PublishRawAsync(id).ConfigureAwait(false)).ConfigureAwait(false); + Assert.IsNotEmpty(initial); + DateTime lastTimestamp = initial.Last().Value.ServerTimestamp; + + await CloseRawSessionAsync().ConfigureAwait(false); + await Task.Delay(TimeSpan.FromSeconds(10)).ConfigureAwait(false); + await MySetUpAsync().ConfigureAwait(false); + await TransferRawAsync(id).ConfigureAwait(false); + + PublishResponse first = await PublishRawAsync(id).ConfigureAwait(false); + Assert.IsNotEmpty(first.NotificationMessage.NotificationData, + "The first Publish after transfer must contain queued data, not a keep-alive."); + List values = await DrainRawAsync(id, first).ConfigureAwait(false); + Assert.Greater(values.Count, 1, "Samples must accumulate while disconnected."); + Assert.GreaterOrEqual(values[0].Value.ServerTimestamp, lastTimestamp); + Assert.IsTrue(values.All(value => StatusCode.IsGood(value.Value.StatusCode))); + // Allow a server clock update (1 second) and scheduling jitter in addition to sampling. + Assert.LessOrEqual((DateTime.UtcNow - values.Last().Value.ServerTimestamp).TotalMilliseconds, + items.Results[0].RevisedSamplingInterval + 2000, + "The final sample must be current relative to the sampling interval."); + } + finally + { + await DeleteRawSubscriptionAsync(id).ConfigureAwait(false); + } + } + + /// + /// Draining queued writes must be followed by a keep-alive without extra initial values. + /// Cover both one response and notification limits requiring MoreNotifications. + /// + [TestCase(0u)] + [TestCase(7u)] + public async Task TransferWithoutInitialValuesReturnsKeepAliveAfterDrainingAsync(uint maxNotifications) + { + const int loopCount = 10; + NodeId[] nodes = GetTestSetStaticMassNumeric(Session.NamespaceUris) + .Where(entry => entry.Value == typeof(int)).Take(3).Select(entry => entry.Key).ToArray(); + Assert.AreEqual(3, nodes.Length); + var originals = new List(); + foreach (NodeId node in nodes) + { + originals.Add((int)(await Session.ReadValueAsync(node).ConfigureAwait(false)).Value); + } + uint id = await CreateRawDurableSubscriptionAsync(maxNotifications).ConfigureAwait(false); + try + { + await WriteRawValuesAsync(nodes, Enumerable.Repeat(-1, nodes.Length).ToArray()).ConfigureAwait(false); + CreateMonitoredItemsResponse items = await CreateRawMonitoredItemsAsync( + id, nodes, loopCount, 100).ConfigureAwait(false); + List initial = await DrainRawAsync(id, + await PublishRawAsync(id).ConfigureAwait(false)).ConfigureAwait(false); + Assert.AreEqual(nodes.Length, initial.Count); + Assert.IsEmpty((await PublishRawAsync(id).ConfigureAwait(false)).NotificationMessage.NotificationData); + + await CloseRawSessionAsync().ConfigureAwait(false); + await MySetUpAsync().ConfigureAwait(false); + double samplingInterval = items.Results.Max(item => item.RevisedSamplingInterval); + for (int value = 0; value < loopCount; value++) + { + await WriteRawValuesAsync(nodes, Enumerable.Repeat(value, nodes.Length).ToArray()).ConfigureAwait(false); + await Task.Delay(TimeSpan.FromMilliseconds(samplingInterval * 1.5)).ConfigureAwait(false); + } + await CloseRawSessionAsync().ConfigureAwait(false); + await MySetUpAsync().ConfigureAwait(false); + await TransferRawAsync(id).ConfigureAwait(false); + + PublishResponse first = await PublishRawAsync(id).ConfigureAwait(false); + Assert.IsNotEmpty(first.NotificationMessage.NotificationData, + "Queued writes must be delivered on the first Publish after transfer."); + if (maxNotifications == 0) + { + Assert.AreEqual(nodes.Length * loopCount, + ((DataChangeNotification)first.NotificationMessage.NotificationData.Single().Body).MonitoredItems.Count); + } + else + { + Assert.IsTrue(first.MoreNotifications); + } + List values = await DrainRawAsync(id, first).ConfigureAwait(false); + Assert.AreEqual(nodes.Length * loopCount, values.Count); + for (uint handle = 1; handle <= nodes.Length; handle++) + { + Assert.AreEqual(Enumerable.Range(0, loopCount), values + .Where(value => value.ClientHandle == handle).Select(value => value.Value.Value), + "Every queued write must be delivered once and in order for each item."); + } + PublishResponse keepAlive = await PublishRawAsync(id).ConfigureAwait(false); + Assert.IsEmpty(keepAlive.NotificationMessage.NotificationData, + "The next Publish after draining must be a keep-alive."); + Assert.IsFalse(keepAlive.MoreNotifications); + } + finally + { + await DeleteRawSubscriptionAsync(id).ConfigureAwait(false); + await WriteRawValuesAsync(nodes, originals.ToArray()).ConfigureAwait(false); + } + } + + private async Task CreateRawDurableSubscriptionAsync(uint maxNotifications) + { + CreateSubscriptionResponse response = await Session.CreateSubscriptionAsync( + null, 100, 1000, 5, maxNotifications, true, 0, CancellationToken.None).ConfigureAwait(false); + IList result = await Session.CallAsync(ObjectIds.Server, + MethodIds.Server_SetSubscriptionDurable, CancellationToken.None, response.SubscriptionId, 1u) + .ConfigureAwait(false); + Assert.AreEqual(1u, result[0]); + return response.SubscriptionId; + } + + private async Task CreateRawMonitoredItemsAsync( + uint id, NodeId[] nodes, uint queueSize, double samplingInterval) + { + var requests = new MonitoredItemCreateRequestCollection(nodes.Select((node, index) => + new MonitoredItemCreateRequest + { + ItemToMonitor = new ReadValueId { NodeId = node, AttributeId = Attributes.Value }, + MonitoringMode = MonitoringMode.Reporting, + RequestedParameters = new MonitoringParameters + { + ClientHandle = (uint)index + 1, + SamplingInterval = samplingInterval, + QueueSize = queueSize, + DiscardOldest = true + } + })); + CreateMonitoredItemsResponse response = await Session.CreateMonitoredItemsAsync( + null, id, TimestampsToReturn.Both, requests, CancellationToken.None).ConfigureAwait(false); + Assert.AreEqual(nodes.Length, response.Results.Count); + foreach (MonitoredItemCreateResult result in response.Results) + { + Assert.IsTrue(StatusCode.IsGood(result.StatusCode)); + Assert.AreEqual(queueSize, result.RevisedQueueSize); + } + return response; + } + + private async Task CloseRawSessionAsync() + { + Session.DeleteSubscriptionsOnClose = false; + await Session.CloseAsync().ConfigureAwait(false); + Session.Dispose(); + Session = null; + } + + private async Task TransferRawAsync(uint id) + { + TransferSubscriptionsResponse response = await Session.TransferSubscriptionsAsync( + null, [id], false, CancellationToken.None).ConfigureAwait(false); + Assert.AreEqual(1, response.Results.Count); + Assert.AreEqual(StatusCodes.Good, response.Results[0].StatusCode.Code); + } + + private async Task PublishRawAsync(uint id) + { + PublishResponse response = await Session.PublishAsync( + new RequestHeader { TimeoutHint = 10000 }, [], CancellationToken.None).ConfigureAwait(false); + Assert.AreEqual(id, response.SubscriptionId); + Assert.AreEqual(StatusCodes.Good, response.ResponseHeader.ServiceResult.Code); + return response; + } + + private async Task> DrainRawAsync(uint id, PublishResponse response) + { + var values = new List(); + for (int responses = 0; ; responses++) + { + Assert.Less(responses, 100, "MoreNotifications must eventually become false."); + foreach (ExtensionObject notification in response.NotificationMessage.NotificationData) + { + Assert.IsInstanceOf(notification.Body); + values.AddRange(((DataChangeNotification)notification.Body).MonitoredItems); + } + if (!response.MoreNotifications) + { + return values; + } + response = await PublishRawAsync(id).ConfigureAwait(false); + } + } + + private async Task WriteRawValuesAsync(NodeId[] nodes, int[] values) + { + var writes = new WriteValueCollection(nodes.Select((node, index) => new WriteValue + { + NodeId = node, + AttributeId = Attributes.Value, + Value = new DataValue(new Variant(values[index])) + })); + WriteResponse response = await Session.WriteAsync(null, writes, CancellationToken.None).ConfigureAwait(false); + Assert.AreEqual(nodes.Length, response.Results.Count); + Assert.IsTrue(response.Results.All(StatusCode.IsGood)); + } + + private async Task DeleteRawSubscriptionAsync(uint id) + { + if (Session == null) + { + await MySetUpAsync().ConfigureAwait(false); + } + // Reclaim the subscription if the test failed while it was abandoned. + await Session.TransferSubscriptionsAsync(null, [id], false, CancellationToken.None).ConfigureAwait(false); + DeleteSubscriptionsResponse response = await Session.DeleteSubscriptionsAsync( + null, [id], CancellationToken.None).ConfigureAwait(false); + Assert.AreEqual(StatusCodes.Good, response.Results[0].Code); + } + private async Task CreateDurableSubscriptionAsync() { var subscription = new TestableSubscription(Session.DefaultSubscription) diff --git a/Tests/Opc.Ua.Server.Tests/SubscriptionTests.cs b/Tests/Opc.Ua.Server.Tests/SubscriptionTests.cs index cb79c8c582..7825dae382 100644 --- a/Tests/Opc.Ua.Server.Tests/SubscriptionTests.cs +++ b/Tests/Opc.Ua.Server.Tests/SubscriptionTests.cs @@ -565,6 +565,60 @@ public async Task ModifySubscriptionWithNotificationLimitsUsesEffectiveLimitAsyn Is.EqualTo((uint)expectedLimit)); } + [TestCase(false)] + [TestCase(true)] + public async Task TransferWithoutInitialValuesChecksQueuedDataBeforeKeepAliveAsync(bool hasQueuedData) + { + m_queueFactoryMock.SetupGet(factory => factory.SupportsDurableQueues).Returns(true); + using Subscription subscription = CreateSubscription(); + Assert.That(ServiceResult.IsGood(subscription.SetSubscriptionDurable(1000)), Is.True); + var publishLimits = new List(); + Mock item = CreateDataChangeMonitoredItem(1, 3, publishLimits); + item.SetupGet(value => value.IsReadyToPublish) + .Returns(() => hasQueuedData && publishLimits.Count == 0); + await RegisterMonitoredItemsAsync(subscription, item.Object).ConfigureAwait(false); + + subscription.SessionClosed(); + SetExpiryTime(subscription, HiResClock.TickCount64 - 100); + Assert.That(subscription.PublishTimerExpired(), Is.EqualTo(PublishingState.NotificationsAvailable)); + Assert.That(GetItemsToPublishCount(subscription), Is.Zero); + SetExpiryTime(subscription, HiResClock.TickCount64 - 100); + subscription.PublishTimerExpired(); + Assert.That(subscription.Diagnostics.CurrentLifetimeCount, Is.GreaterThan(0)); + + // Transfer just after an abandoned subscription's timer tick. The next queue scan + // must not depend on another publishing interval elapsing. + SetExpiryTime(subscription, HiResClock.TickCount64 + 100000); + var newSession = new Mock(); + newSession.SetupGet(session => session.Id).Returns(new NodeId(Guid.NewGuid())); + var context = new OperationContext(newSession.Object, new DiagnosticsMasks()); + await subscription.TransferSessionAsync(context, false).ConfigureAwait(false); + Assert.That(subscription.Diagnostics.CurrentLifetimeCount, Is.Zero); + Assert.That(subscription.PublishTimerExpired(), Is.Not.EqualTo(PublishingState.Idle)); + + NotificationMessage first = subscription.Publish(context, out _, out bool moreNotifications); + Assert.That(first, Is.Not.Null); + Assert.That(first.NotificationData, Has.Count.EqualTo(hasQueuedData ? 1 : 0)); + Assert.That(moreNotifications, Is.False); + if (hasQueuedData) + { + var data = (DataChangeNotification)first.NotificationData[0].Body; + Assert.That(data.MonitoredItems.Select(value => value.Value.Value), Is.EqualTo(Enumerable.Range(0, 3))); + } + item.Verify(value => value.SetupResendDataTrigger(), Times.Never); + + // No new samples: the next keep-alive must not replay the values just drained. + for (int tick = 0; tick < 5; tick++) + { + SetExpiryTime(subscription, HiResClock.TickCount64 - 100); + subscription.PublishTimerExpired(); + } + NotificationMessage keepAlive = subscription.Publish(context, out _, out moreNotifications); + Assert.That(keepAlive, Is.Not.Null); + Assert.That(keepAlive.NotificationData, Is.Empty); + Assert.That(moreNotifications, Is.False); + } + private async Task RegisterMonitoredItemsAsync( Subscription subscription, params IMonitoredItem[] monitoredItems) From 36acf1bb793a8ce5fba2b1cdcd8efd674d6f1fa6 Mon Sep 17 00:00:00 2001 From: romanett Date: Fri, 2 Oct 2026 07:24:37 +0200 Subject: [PATCH 6/6] [master378] Fix GDS client deadlock after a failed keep-alive (#4623) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Proposed changes Minimal backport of the `GlobalDiscoveryServerClient` keep-alive part of #4158 to `master378`. On a bad keep-alive, `Session_KeepAliveAsync` took `m_lock` and called `Session.Dispose()` on the keep-alive worker. `Session.Dispose()` waits for that worker to stop, so the client deadlocked while holding `m_lock`, and every later call (for example `FindApplicationAsync`) hung. The handler now returns right away and runs the cleanup with `Task.Run`, so the keep-alive worker can stop before the session is disposed. The cleanup skips its work if the client was disposed in the meantime (it handles `ObjectDisposedException` on the lock and checks `m_disposed`). `BackgroundTaskScope` from #4158 is not on 1.5.378, so this uses a plain `Task.Run`. No public API changes. ## Related Issues - Fixes #4621 ## Types of changes - [x] Bugfix (non-breaking change which fixes an issue) ## Checklist - [x] Existing GDS client tests pass locally (`Opc.Ua.Gds.Tests` `ClientTest`, net10.0: 102/102) 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../GlobalDiscoveryServerClient.cs | 42 +++++++++++++++---- 1 file changed, 34 insertions(+), 8 deletions(-) diff --git a/Libraries/Opc.Ua.Gds.Client.Common/GlobalDiscoveryServerClient.cs b/Libraries/Opc.Ua.Gds.Client.Common/GlobalDiscoveryServerClient.cs index 634e3b0210..22528996ae 100644 --- a/Libraries/Opc.Ua.Gds.Client.Common/GlobalDiscoveryServerClient.cs +++ b/Libraries/Opc.Ua.Gds.Client.Common/GlobalDiscoveryServerClient.cs @@ -490,17 +490,35 @@ public async Task DisconnectAsync(CancellationToken ct = default) } } - private async void Session_KeepAliveAsync(ISession session, KeepAliveEventArgs e) + private void Session_KeepAlive(ISession session, KeepAliveEventArgs e) { - if (ServiceResult.IsBad(e.Status) && !m_disposed) + if (!ServiceResult.IsBad(e.Status) || m_disposed) { - await m_lock.WaitAsync().ConfigureAwait(false); + return; + } + + // Disposing the session waits for its keep-alive worker to stop, and + // this callback runs on that worker. Move the cleanup off the worker + // so the callback returns first, otherwise the dispose deadlocks while + // holding m_lock and every later call on this client hangs. + _ = Task.Run(async () => + { + try + { + await m_lock.WaitAsync().ConfigureAwait(false); + } + catch (ObjectDisposedException) + { + // the client was disposed, which also disposed the session. + return; + } + try { - if (session == Session) + if (!m_disposed && session == Session) { - Session.Dispose(); Session = null; + session.Dispose(); } } catch (Exception ex) @@ -509,9 +527,17 @@ private async void Session_KeepAliveAsync(ISession session, KeepAliveEventArgs e } finally { - m_lock.Release(); + try + { + m_lock.Release(); + } + catch (ObjectDisposedException) + { + // Dispose handed the lock to this task on release and + // then disposed it. + } } - } + }); } private void Session_SessionClosing(object sender, EventArgs e) @@ -1560,7 +1586,7 @@ private async Task ConnectInternalAsync( .ConfigureAwait(false); Session.SessionClosing += Session_SessionClosing; - Session.KeepAlive += Session_KeepAliveAsync; + Session.KeepAlive += Session_KeepAlive; Session.KeepAlive += KeepAlive; // TODO: implement, suppress warning/error