mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-22 01:26:20 +08:00
Marc Schier
fde50c5700
# Description This draft isolates the Session/ActivateSession, reconnect, and distributed Session security work split from #4021. ### Failure - `ActivateSession` could transfer a Session to a new SecureChannel without consistently enforcing the original client certificate, MessageSecurityMode, SecurityPolicy, and exact OPC `ClientUserId` semantics. - Missing or legacy distributed Session security state did not provide enough information to validate transfer safely. - Malformed or reused nonces, concurrent activation, and cancellation could leave activation state inconsistent. - Client reconnect could treat recoverable Session-loss/security responses as fatal, reuse a Session after client-certificate rotation, or report a managed channel ready before Session recreation completed. ### Fix - Serialize activation, consume server nonces once, validate application signatures and user tokens asynchronously, and enforce transfer continuity against the original channel and authenticated identity. - Persist versioned transfer-security state for distributed Sessions and fail closed when certificate, identity, activation, or nonce state is absent or invalid. - Validate server signatures/nonces on the client, recreate Sessions for certificate rotation and recoverable activation failures, and await participant recreation before transitioning the managed channel to `Ready`. - Make the restart recovery integration test use bounded reconnect cycles instead of a fixed startup delay. ### Specification - OPC 10000-4 §5.7.2.2 — CreateSession signatures and nonces. - OPC 10000-4 §5.7.3.1 — ActivateSession identity continuity and SecureChannel transfer. - OPC 10000-4 §6.1.8 — application nonce generation, validation, and reuse requirements. ### Validation - `git diff --check` - Release `net10.0`: `Opc.Ua.Redundancy.Server.Tests` 567/567 plus the targeted server Session, Subscription transfer, ClientUserId, client reconnect and certificate rotation suites pass. - Release `net48`: the same redundancy suite (567/567) and the targeted Session/Subscription suites pass. - Full `Opc.Ua.Server.Tests` run on `net10.0`: 3633/3634. The single failure, `ServerFluentApiHostingTests.ConfigureApplicationBuildsSharedClientAndServerConfigurationAsync`, reproduces identically on the unmodified commit and is a local certificate-store path issue rather than a regression. These were targeted validation runs; no long-soak result is claimed.
874 lines
34 KiB
C#
874 lines
34 KiB
C#
/* ========================================================================
|
|
* 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/
|
|
* ======================================================================*/
|
|
|
|
#nullable enable
|
|
|
|
using System;
|
|
using System.Collections.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Moq;
|
|
using NUnit.Framework;
|
|
using Opc.Ua.Bindings;
|
|
using Opc.Ua.Security.Certificates;
|
|
using Opc.Ua.Tests;
|
|
|
|
// CA2000: test code; disposables are released by test cleanup paths.
|
|
#pragma warning disable CA2000
|
|
|
|
namespace Opc.Ua.Core.Tests.Stack.Client
|
|
{
|
|
/// <summary>
|
|
/// Tests certificate-rotation integration for <see cref="ClientChannelManager"/>.
|
|
/// </summary>
|
|
[TestFixture]
|
|
[Category("Client")]
|
|
[SetCulture("en-us")]
|
|
[SetUICulture("en-us")]
|
|
[Parallelizable]
|
|
public sealed class ClientChannelManagerCertRotationTests
|
|
{
|
|
private static readonly ICertificateFactory s_factory = DefaultCertificateFactory.Instance;
|
|
private static readonly int[] s_firstTwoReconnectAttempts = [0, 1];
|
|
|
|
[Test]
|
|
public async Task CertificateRotationTriggersReconnectAllAsync()
|
|
{
|
|
using Certificate oldCertificate = s_factory.CreateCertificate("CN=old-client").CreateForRSA();
|
|
using Certificate newCertificate = s_factory.CreateCertificate("CN=new-client").CreateForRSA();
|
|
using Certificate serverCertificate = s_factory.CreateCertificate("CN=server").CreateForRSA();
|
|
|
|
TestCertificateChangeSource changes = new();
|
|
ConcurrentQueue<TransportChannelSettings> openSettings = new();
|
|
ClientChannelManager sut = CreateSut(oldCertificate, changes, openSettings);
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCertificate);
|
|
TestParticipant firstParticipant = new("first", endpoint);
|
|
TestParticipant secondParticipant = new("second", endpoint);
|
|
IManagedTransportChannel? firstChannel = null;
|
|
IManagedTransportChannel? secondChannel = null;
|
|
|
|
try
|
|
{
|
|
sut.UpdateClientCertificate(oldCertificate, null);
|
|
firstChannel = await sut.GetAsync(firstParticipant).ConfigureAwait(false);
|
|
secondChannel = await sut.GetAsync(secondParticipant).ConfigureAwait(false);
|
|
|
|
changes.Raise(new CertificateChangeEvent(
|
|
CertificateChangeKind.ApplicationCertificateUpdated,
|
|
TrustListIdentifier.Peers,
|
|
ObjectTypeIds.RsaSha256ApplicationCertificateType,
|
|
oldCertificate,
|
|
newCertificate,
|
|
null));
|
|
|
|
await WaitUntilAsync(() =>
|
|
firstParticipant.NotificationCount == 1 &&
|
|
secondParticipant.NotificationCount == 1 &&
|
|
openSettings.Count >= 2).ConfigureAwait(false);
|
|
|
|
TransportChannelSettings[] settings = [.. openSettings];
|
|
Assert.That(settings[^1].ClientCertificate?.Thumbprint, Is.EqualTo(newCertificate.Thumbprint));
|
|
Assert.That(firstParticipant.NotificationCount, Is.EqualTo(1));
|
|
Assert.That(secondParticipant.NotificationCount, Is.EqualTo(1));
|
|
}
|
|
finally
|
|
{
|
|
if (firstChannel != null)
|
|
{
|
|
await firstChannel.CloseAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
if (secondChannel != null)
|
|
{
|
|
await secondChannel.CloseAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
|
|
// The production channel factory parses description.ServerCertificate
|
|
// into a Certificate that the real channel would own and dispose; the
|
|
// mock channel does not, so dispose the captured copies here.
|
|
while (openSettings.TryDequeue(out TransportChannelSettings? opened))
|
|
{
|
|
opened.ServerCertificate?.Dispose();
|
|
}
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task CertificateRotationIgnoresUnrelatedCerts()
|
|
{
|
|
using Certificate oldCertificate = s_factory.CreateCertificate("CN=old-client").CreateForRSA();
|
|
using Certificate unrelatedOldCertificate = s_factory.CreateCertificate("CN=old-other").CreateForRSA();
|
|
using Certificate unrelatedNewCertificate = s_factory.CreateCertificate("CN=new-other").CreateForRSA();
|
|
using Certificate serverCertificate = s_factory.CreateCertificate("CN=server").CreateForRSA();
|
|
|
|
TestCertificateChangeSource changes = new();
|
|
ConcurrentQueue<TransportChannelSettings> openSettings = new();
|
|
ClientChannelManager sut = CreateSut(oldCertificate, changes, openSettings);
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCertificate);
|
|
TestParticipant participant = new("participant", endpoint);
|
|
IManagedTransportChannel? channel = null;
|
|
|
|
try
|
|
{
|
|
sut.UpdateClientCertificate(oldCertificate, null);
|
|
channel = await sut.GetAsync(participant).ConfigureAwait(false);
|
|
|
|
changes.Raise(new CertificateChangeEvent(
|
|
CertificateChangeKind.ApplicationCertificateUpdated,
|
|
TrustListIdentifier.Peers,
|
|
ObjectTypeIds.RsaSha256ApplicationCertificateType,
|
|
unrelatedOldCertificate,
|
|
unrelatedNewCertificate,
|
|
null));
|
|
|
|
await Task.Delay(TimeSpan.FromMilliseconds(100)).ConfigureAwait(false);
|
|
|
|
Assert.That(participant.NotificationCount, Is.Zero);
|
|
Assert.That(openSettings, Has.Count.EqualTo(1));
|
|
}
|
|
finally
|
|
{
|
|
if (channel != null)
|
|
{
|
|
await channel.CloseAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
|
|
// The production channel factory parses description.ServerCertificate
|
|
// into a Certificate that the real channel would own and dispose; the
|
|
// mock channel does not, so dispose the captured copies here.
|
|
while (openSettings.TryDequeue(out TransportChannelSettings? opened))
|
|
{
|
|
opened.ServerCertificate?.Dispose();
|
|
}
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectWaitsForRequiredParticipantRecreationAsync()
|
|
{
|
|
using Certificate clientCertificate = s_factory
|
|
.CreateCertificate("CN=recreate-client")
|
|
.CreateForRSA();
|
|
using Certificate serverCertificate = s_factory
|
|
.CreateCertificate("CN=recreate-server")
|
|
.CreateForRSA();
|
|
|
|
TestCertificateChangeSource changes = new();
|
|
ConcurrentQueue<TransportChannelSettings> openSettings = new();
|
|
ClientChannelManager sut = CreateSut(
|
|
clientCertificate,
|
|
changes,
|
|
openSettings);
|
|
var participant = new BlockingRecreateParticipant(
|
|
"recreate",
|
|
GetTestEndpoint(serverCertificate));
|
|
IManagedTransportChannel? channel = null;
|
|
|
|
try
|
|
{
|
|
sut.UpdateClientCertificate(clientCertificate, null);
|
|
channel = await sut.GetAsync(participant).ConfigureAwait(false);
|
|
|
|
Task reconnect = sut.ReconnectAsync(channel).AsTask();
|
|
await participant.RecreateStarted.Task
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(reconnect.IsCompleted, Is.False);
|
|
Assert.That(channel.State, Is.Not.EqualTo(ChannelState.Ready));
|
|
|
|
participant.ReleaseRecreate();
|
|
await reconnect.ConfigureAwait(false);
|
|
|
|
Assert.That(participant.RecreateCount, Is.EqualTo(1));
|
|
Assert.That(channel.State, Is.EqualTo(ChannelState.Ready));
|
|
}
|
|
finally
|
|
{
|
|
participant.ReleaseRecreate();
|
|
if (channel != null)
|
|
{
|
|
await channel.CloseAsync().ConfigureAwait(false);
|
|
}
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
while (openSettings.TryDequeue(out TransportChannelSettings? opened))
|
|
{
|
|
opened.ServerCertificate?.Dispose();
|
|
}
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectRetriesAfterParticipantRecreationFailureAsync()
|
|
{
|
|
using Certificate clientCertificate = s_factory
|
|
.CreateCertificate("CN=retry-client")
|
|
.CreateForRSA();
|
|
using Certificate serverCertificate = s_factory
|
|
.CreateCertificate("CN=retry-server")
|
|
.CreateForRSA();
|
|
|
|
TestCertificateChangeSource changes = new();
|
|
ConcurrentQueue<TransportChannelSettings> openSettings = new();
|
|
ClientChannelManager sut = CreateSut(
|
|
clientCertificate,
|
|
changes,
|
|
openSettings,
|
|
new FixedAttemptReconnectPolicy(2));
|
|
var participant = new FailingRecreateParticipant(
|
|
"retry",
|
|
GetTestEndpoint(serverCertificate),
|
|
failuresBeforeSuccess: 1);
|
|
var states = new ConcurrentQueue<ChannelState>();
|
|
IManagedTransportChannel? channel = null;
|
|
|
|
try
|
|
{
|
|
sut.UpdateClientCertificate(clientCertificate, null);
|
|
channel = await sut.GetAsync(participant).ConfigureAwait(false);
|
|
channel.StateChanged += (_, change) => states.Enqueue(change.NewState);
|
|
|
|
await sut.ReconnectAsync(channel).ConfigureAwait(false);
|
|
|
|
Assert.That(participant.RecreateCount, Is.EqualTo(2));
|
|
Assert.That(participant.ReconnectAttempts, Is.EqualTo(s_firstTwoReconnectAttempts));
|
|
Assert.That(
|
|
states.Count(state => state == ChannelState.TransportReconnecting),
|
|
Is.EqualTo(2));
|
|
Assert.That(
|
|
states.Count(state => state == ChannelState.TransportConnectedSessionReactivating),
|
|
Is.EqualTo(2));
|
|
Assert.That(channel.State, Is.EqualTo(ChannelState.Ready));
|
|
}
|
|
finally
|
|
{
|
|
if (channel != null)
|
|
{
|
|
await channel.CloseAsync().ConfigureAwait(false);
|
|
}
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
DisposeOpenedServerCertificates(openSettings);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectFaultsAfterParticipantRecreationFailuresExhaustPolicyAsync()
|
|
{
|
|
using Certificate clientCertificate = s_factory
|
|
.CreateCertificate("CN=failure-client")
|
|
.CreateForRSA();
|
|
using Certificate serverCertificate = s_factory
|
|
.CreateCertificate("CN=failure-server")
|
|
.CreateForRSA();
|
|
|
|
TestCertificateChangeSource changes = new();
|
|
ConcurrentQueue<TransportChannelSettings> openSettings = new();
|
|
ClientChannelManager sut = CreateSut(
|
|
clientCertificate,
|
|
changes,
|
|
openSettings,
|
|
new FixedAttemptReconnectPolicy(2));
|
|
var participant = new FailingRecreateParticipant(
|
|
"failure",
|
|
GetTestEndpoint(serverCertificate),
|
|
failuresBeforeSuccess: int.MaxValue);
|
|
IManagedTransportChannel? channel = null;
|
|
|
|
try
|
|
{
|
|
sut.UpdateClientCertificate(clientCertificate, null);
|
|
channel = await sut.GetAsync(participant).ConfigureAwait(false);
|
|
|
|
await sut.ReconnectAsync(channel).ConfigureAwait(false);
|
|
|
|
Assert.That(participant.RecreateCount, Is.EqualTo(2));
|
|
Assert.That(participant.ReconnectAttempts, Is.EqualTo(s_firstTwoReconnectAttempts));
|
|
Assert.That(channel.State, Is.EqualTo(ChannelState.Faulted));
|
|
Assert.That(
|
|
sut.GetChannelDiagnostics().Single().LastReconnectAttempt,
|
|
Is.EqualTo(2));
|
|
}
|
|
finally
|
|
{
|
|
if (channel != null)
|
|
{
|
|
await channel.CloseAsync().ConfigureAwait(false);
|
|
}
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
DisposeOpenedServerCertificates(openSettings);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposeCancelsParticipantRecreationAndCleansEntryAsync()
|
|
{
|
|
using Certificate clientCertificate = s_factory
|
|
.CreateCertificate("CN=cancel-client")
|
|
.CreateForRSA();
|
|
using Certificate serverCertificate = s_factory
|
|
.CreateCertificate("CN=cancel-server")
|
|
.CreateForRSA();
|
|
|
|
TestCertificateChangeSource changes = new();
|
|
ConcurrentQueue<TransportChannelSettings> openSettings = new();
|
|
ClientChannelManager sut = CreateSut(
|
|
clientCertificate,
|
|
changes,
|
|
openSettings,
|
|
new FixedAttemptReconnectPolicy(3));
|
|
var participant = new CancellationAwareRecreateParticipant(
|
|
"cancel",
|
|
GetTestEndpoint(serverCertificate));
|
|
IManagedTransportChannel? channel = null;
|
|
|
|
try
|
|
{
|
|
channel = await sut.GetAsync(participant).ConfigureAwait(false);
|
|
Task reconnect = sut.ReconnectAsync(channel).AsTask();
|
|
await participant.RecreateStarted.Task
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
await participant.RecreateCanceled.Task
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(
|
|
async () => await reconnect.ConfigureAwait(false),
|
|
Throws.InstanceOf<OperationCanceledException>());
|
|
Assert.That(participant.RecreateCount, Is.EqualTo(1));
|
|
Assert.That(channel.State, Is.EqualTo(ChannelState.Closed));
|
|
Assert.That(sut.GetChannelDiagnostics(), Is.Empty);
|
|
}
|
|
finally
|
|
{
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
DisposeOpenedServerCertificates(openSettings);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectingStaleLeasesReusesReplacementEntryAsync()
|
|
{
|
|
using Certificate clientCertificate = s_factory
|
|
.CreateCertificate("CN=stale-client")
|
|
.CreateForRSA();
|
|
using Certificate serverCertificate = s_factory
|
|
.CreateCertificate("CN=stale-server")
|
|
.CreateForRSA();
|
|
|
|
TestCertificateChangeSource changes = new();
|
|
ConcurrentQueue<TransportChannelSettings> openSettings = new();
|
|
ClientChannelManager sut = CreateSut(clientCertificate, changes, openSettings);
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(serverCertificate);
|
|
var fatalOnce = new FatalOnceParticipant("fatal-once", endpoint);
|
|
var stable = new TestParticipant("stable", endpoint);
|
|
IManagedTransportChannel? firstChannel = null;
|
|
IManagedTransportChannel? secondChannel = null;
|
|
|
|
try
|
|
{
|
|
sut.UpdateClientCertificate(clientCertificate, null);
|
|
firstChannel = await sut.GetAsync(fatalOnce).ConfigureAwait(false);
|
|
secondChannel = await sut.GetAsync(stable).ConfigureAwait(false);
|
|
|
|
await sut.ReconnectAsync(firstChannel).ConfigureAwait(false);
|
|
Assert.That(firstChannel.State, Is.EqualTo(ChannelState.Faulted));
|
|
Assert.That(secondChannel.State, Is.EqualTo(ChannelState.Faulted));
|
|
|
|
await sut.ReconnectAsync(firstChannel).ConfigureAwait(false);
|
|
await sut.ReconnectAsync(secondChannel).ConfigureAwait(false);
|
|
|
|
ManagedChannelDiagnostic diagnostic = sut.GetChannelDiagnostics().Single();
|
|
Assert.That(firstChannel.State, Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(secondChannel.State, Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(diagnostic.Refcount, Is.EqualTo(2));
|
|
Assert.That(diagnostic.ParticipantCount, Is.EqualTo(2));
|
|
}
|
|
finally
|
|
{
|
|
if (firstChannel != null)
|
|
{
|
|
await firstChannel.CloseAsync().ConfigureAwait(false);
|
|
}
|
|
if (secondChannel != null)
|
|
{
|
|
await secondChannel.CloseAsync().ConfigureAwait(false);
|
|
}
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
DisposeOpenedServerCertificates(openSettings);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposeUnsubscribesFromCertEvent()
|
|
{
|
|
using Certificate oldCertificate = s_factory.CreateCertificate("CN=old-client").CreateForRSA();
|
|
using Certificate newCertificate = s_factory.CreateCertificate("CN=new-client").CreateForRSA();
|
|
|
|
TestCertificateChangeSource changes = new();
|
|
ConcurrentQueue<TransportChannelSettings> openSettings = new();
|
|
ClientChannelManager sut = CreateSut(oldCertificate, changes, openSettings);
|
|
|
|
Assert.That(changes.ObserverCount, Is.EqualTo(1));
|
|
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
changes.Raise(new CertificateChangeEvent(
|
|
CertificateChangeKind.ApplicationCertificateUpdated,
|
|
TrustListIdentifier.Peers,
|
|
ObjectTypeIds.RsaSha256ApplicationCertificateType,
|
|
oldCertificate,
|
|
newCertificate,
|
|
null));
|
|
|
|
await Task.Delay(TimeSpan.FromMilliseconds(100)).ConfigureAwait(false);
|
|
|
|
Assert.That(changes.ObserverCount, Is.Zero);
|
|
Assert.That(openSettings, Is.Empty);
|
|
}
|
|
|
|
private static ClientChannelManager CreateSut(
|
|
Certificate applicationCertificate,
|
|
TestCertificateChangeSource changes,
|
|
ConcurrentQueue<TransportChannelSettings> openSettings,
|
|
IChannelReconnectPolicy? reconnectPolicy = null)
|
|
{
|
|
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
|
|
var certificateManager = new Mock<ICertificateManager>();
|
|
certificateManager.SetupGet(m => m.CertificateChanges).Returns(changes);
|
|
|
|
var channel = new Mock<IChannel>();
|
|
channel.Setup(c => c.OpenAsync(
|
|
It.IsAny<Uri>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback<Uri, TransportChannelSettings, CancellationToken>((_, settings, _) =>
|
|
openSettings.Enqueue(settings))
|
|
.Returns(new ValueTask());
|
|
channel.Setup(c => c.OpenAsync(
|
|
It.IsAny<ITransportWaitingConnection>(),
|
|
It.IsAny<TransportChannelSettings>(),
|
|
It.IsAny<CancellationToken>()))
|
|
.Callback<ITransportWaitingConnection, TransportChannelSettings, CancellationToken>((_, settings, _) =>
|
|
openSettings.Enqueue(settings))
|
|
.Returns(new ValueTask());
|
|
channel.Setup(c => c.CloseAsync(It.IsAny<CancellationToken>())).Returns(new ValueTask());
|
|
channel.Setup(c => c.SupportedFeatures).Returns(TransportChannelFeatures.None);
|
|
|
|
var bindings = new Mock<ITransportChannelBindings>();
|
|
bindings.Setup(b => b.Create(It.IsAny<string>(), It.IsAny<ITelemetryContext>()))
|
|
.Returns(channel.Object);
|
|
|
|
var configuration = new ApplicationConfiguration(telemetry)
|
|
{
|
|
CertificateManager = certificateManager.Object
|
|
};
|
|
configuration.SecurityConfiguration.ApplicationCertificate = new CertificateIdentifier
|
|
{
|
|
CertificateType = ObjectTypeIds.RsaSha256ApplicationCertificateType,
|
|
Thumbprint = applicationCertificate.Thumbprint
|
|
};
|
|
configuration.SecurityConfiguration.SendCertificateChain = false;
|
|
|
|
return new ClientChannelManager(
|
|
configuration,
|
|
telemetry,
|
|
bindings.Object,
|
|
reconnectPolicy ?? new ImmediateReconnectPolicy());
|
|
}
|
|
|
|
private static ConfiguredEndpoint GetTestEndpoint(Certificate serverCertificate)
|
|
{
|
|
var endpoint = new ConfiguredEndpoint
|
|
{
|
|
Configuration = new EndpointConfiguration
|
|
{
|
|
OperationTimeout = 6000
|
|
}
|
|
};
|
|
endpoint.Description.EndpointUrl = "opc.tcp://localhost:4840";
|
|
endpoint.Description.SecurityMode = MessageSecurityMode.SignAndEncrypt;
|
|
endpoint.Description.SecurityPolicyUri = SecurityPolicies.Basic256Sha256;
|
|
endpoint.Description.ServerCertificate = serverCertificate.RawData.ToByteString();
|
|
return endpoint;
|
|
}
|
|
|
|
private static async Task WaitUntilAsync(Func<bool> condition)
|
|
{
|
|
DateTime deadline = DateTime.UtcNow.AddSeconds(5);
|
|
while (!condition())
|
|
{
|
|
if (DateTime.UtcNow >= deadline)
|
|
{
|
|
Assert.Fail("The certificate rotation reconnect did not complete in time.");
|
|
}
|
|
|
|
await Task.Delay(TimeSpan.FromMilliseconds(20)).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
private static void DisposeOpenedServerCertificates(
|
|
ConcurrentQueue<TransportChannelSettings> openSettings)
|
|
{
|
|
while (openSettings.TryDequeue(out TransportChannelSettings? opened))
|
|
{
|
|
opened.ServerCertificate?.Dispose();
|
|
}
|
|
}
|
|
|
|
public interface IChannel : ITransportChannel, ISecureChannel;
|
|
|
|
private sealed class ImmediateReconnectPolicy : IChannelReconnectPolicy
|
|
{
|
|
public TimeSpan GetDelay(int attempt)
|
|
{
|
|
return attempt == 0 ? TimeSpan.Zero : Timeout.InfiniteTimeSpan;
|
|
}
|
|
}
|
|
|
|
private sealed class FixedAttemptReconnectPolicy : IChannelReconnectPolicy
|
|
{
|
|
public FixedAttemptReconnectPolicy(int maxAttempts)
|
|
{
|
|
m_maxAttempts = maxAttempts;
|
|
}
|
|
|
|
public TimeSpan GetDelay(int attempt)
|
|
{
|
|
return attempt < m_maxAttempts
|
|
? TimeSpan.Zero
|
|
: Timeout.InfiniteTimeSpan;
|
|
}
|
|
|
|
private readonly int m_maxAttempts;
|
|
}
|
|
|
|
private sealed class TestCertificateChangeSource : IObservable<CertificateChangeEvent>
|
|
{
|
|
public int ObserverCount
|
|
{
|
|
get
|
|
{
|
|
lock (m_lock)
|
|
{
|
|
return m_observers.Count;
|
|
}
|
|
}
|
|
}
|
|
|
|
public IDisposable Subscribe(IObserver<CertificateChangeEvent> observer)
|
|
{
|
|
if (observer == null)
|
|
{
|
|
throw new ArgumentNullException(nameof(observer));
|
|
}
|
|
|
|
lock (m_lock)
|
|
{
|
|
m_observers.Add(observer);
|
|
}
|
|
|
|
return new Subscription(this, observer);
|
|
}
|
|
|
|
public void Raise(CertificateChangeEvent evt)
|
|
{
|
|
IObserver<CertificateChangeEvent>[] snapshot;
|
|
lock (m_lock)
|
|
{
|
|
snapshot = [.. m_observers];
|
|
}
|
|
|
|
for (int i = 0; i < snapshot.Length; i++)
|
|
{
|
|
snapshot[i].OnNext(evt);
|
|
}
|
|
}
|
|
|
|
private void Unsubscribe(IObserver<CertificateChangeEvent> observer)
|
|
{
|
|
lock (m_lock)
|
|
{
|
|
m_observers.Remove(observer);
|
|
}
|
|
}
|
|
|
|
private sealed class Subscription : IDisposable
|
|
{
|
|
public Subscription(
|
|
TestCertificateChangeSource owner,
|
|
IObserver<CertificateChangeEvent> observer)
|
|
{
|
|
m_owner = owner;
|
|
m_observer = observer;
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
if (Interlocked.Exchange(ref m_disposed, 1) != 0)
|
|
{
|
|
return;
|
|
}
|
|
|
|
m_owner.Unsubscribe(m_observer);
|
|
}
|
|
|
|
private readonly TestCertificateChangeSource m_owner;
|
|
private readonly IObserver<CertificateChangeEvent> m_observer;
|
|
private int m_disposed;
|
|
}
|
|
|
|
private readonly Lock m_lock = new();
|
|
private readonly List<IObserver<CertificateChangeEvent>> m_observers = [];
|
|
}
|
|
|
|
private sealed class TestParticipant : IReconnectParticipant
|
|
{
|
|
private int m_notificationCount;
|
|
|
|
public TestParticipant(string id, ConfiguredEndpoint endpoint)
|
|
{
|
|
Id = id;
|
|
Endpoint = endpoint;
|
|
}
|
|
|
|
public string Id { get; }
|
|
|
|
public ConfiguredEndpoint Endpoint { get; }
|
|
|
|
public int NotificationCount => Volatile.Read(ref m_notificationCount);
|
|
|
|
public ValueTask<ParticipantReconnectResult> OnReconnectAsync(
|
|
IManagedTransportChannel channel,
|
|
int reconnectAttempt,
|
|
CancellationToken ct)
|
|
{
|
|
Interlocked.Increment(ref m_notificationCount);
|
|
return new ValueTask<ParticipantReconnectResult>(ParticipantReconnectResult.Reactivated);
|
|
}
|
|
}
|
|
|
|
private sealed class BlockingRecreateParticipant :
|
|
IRecreateAwareReconnectParticipant
|
|
{
|
|
public BlockingRecreateParticipant(
|
|
string id,
|
|
ConfiguredEndpoint endpoint)
|
|
{
|
|
Id = id;
|
|
Endpoint = endpoint;
|
|
}
|
|
|
|
public string Id { get; }
|
|
|
|
public ConfiguredEndpoint Endpoint { get; }
|
|
|
|
public TaskCompletionSource<bool> RecreateStarted { get; } =
|
|
new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
|
|
public int RecreateCount => Volatile.Read(ref m_recreateCount);
|
|
|
|
public ValueTask<ParticipantReconnectResult> OnReconnectAsync(
|
|
IManagedTransportChannel channel,
|
|
int reconnectAttempt,
|
|
CancellationToken ct)
|
|
{
|
|
return new ValueTask<ParticipantReconnectResult>(
|
|
reconnectAttempt < 0
|
|
? ParticipantReconnectResult.Reactivated
|
|
: ParticipantReconnectResult.RequiresSessionRecreate);
|
|
}
|
|
|
|
public async ValueTask RecreateAsync(CancellationToken ct = default)
|
|
{
|
|
Interlocked.Increment(ref m_recreateCount);
|
|
RecreateStarted.TrySetResult(true);
|
|
await m_release.Task.WaitAsync(ct).ConfigureAwait(false);
|
|
}
|
|
|
|
public void ReleaseRecreate()
|
|
{
|
|
m_release.TrySetResult(true);
|
|
}
|
|
|
|
private readonly TaskCompletionSource<bool> m_release =
|
|
new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
|
|
private int m_recreateCount;
|
|
}
|
|
|
|
private sealed class FailingRecreateParticipant :
|
|
IRecreateAwareReconnectParticipant
|
|
{
|
|
public FailingRecreateParticipant(
|
|
string id,
|
|
ConfiguredEndpoint endpoint,
|
|
int failuresBeforeSuccess)
|
|
{
|
|
Id = id;
|
|
Endpoint = endpoint;
|
|
m_failuresBeforeSuccess = failuresBeforeSuccess;
|
|
}
|
|
|
|
public string Id { get; }
|
|
|
|
public ConfiguredEndpoint Endpoint { get; }
|
|
|
|
public int RecreateCount => Volatile.Read(ref m_recreateCount);
|
|
|
|
public int[] ReconnectAttempts => [.. m_reconnectAttempts];
|
|
|
|
public ValueTask<ParticipantReconnectResult> OnReconnectAsync(
|
|
IManagedTransportChannel channel,
|
|
int reconnectAttempt,
|
|
CancellationToken ct)
|
|
{
|
|
if (reconnectAttempt < 0)
|
|
{
|
|
return new ValueTask<ParticipantReconnectResult>(
|
|
ParticipantReconnectResult.Reactivated);
|
|
}
|
|
|
|
m_reconnectAttempts.Enqueue(reconnectAttempt);
|
|
return new ValueTask<ParticipantReconnectResult>(
|
|
ParticipantReconnectResult.RequiresSessionRecreate);
|
|
}
|
|
|
|
public ValueTask RecreateAsync(CancellationToken ct = default)
|
|
{
|
|
int attempt = Interlocked.Increment(ref m_recreateCount);
|
|
if (attempt <= m_failuresBeforeSuccess)
|
|
{
|
|
throw new InvalidOperationException("Injected recreation failure.");
|
|
}
|
|
|
|
return default;
|
|
}
|
|
|
|
private readonly ConcurrentQueue<int> m_reconnectAttempts = new();
|
|
private readonly int m_failuresBeforeSuccess;
|
|
private int m_recreateCount;
|
|
}
|
|
|
|
private sealed class CancellationAwareRecreateParticipant :
|
|
IRecreateAwareReconnectParticipant
|
|
{
|
|
public CancellationAwareRecreateParticipant(
|
|
string id,
|
|
ConfiguredEndpoint endpoint)
|
|
{
|
|
Id = id;
|
|
Endpoint = endpoint;
|
|
}
|
|
|
|
public string Id { get; }
|
|
|
|
public ConfiguredEndpoint Endpoint { get; }
|
|
|
|
public int RecreateCount => Volatile.Read(ref m_recreateCount);
|
|
|
|
public TaskCompletionSource<bool> RecreateStarted { get; } =
|
|
new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
|
|
public TaskCompletionSource<bool> RecreateCanceled { get; } =
|
|
new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
|
|
public ValueTask<ParticipantReconnectResult> OnReconnectAsync(
|
|
IManagedTransportChannel channel,
|
|
int reconnectAttempt,
|
|
CancellationToken ct)
|
|
{
|
|
return new ValueTask<ParticipantReconnectResult>(
|
|
reconnectAttempt < 0
|
|
? ParticipantReconnectResult.Reactivated
|
|
: ParticipantReconnectResult.RequiresSessionRecreate);
|
|
}
|
|
|
|
public async ValueTask RecreateAsync(CancellationToken ct = default)
|
|
{
|
|
Interlocked.Increment(ref m_recreateCount);
|
|
RecreateStarted.TrySetResult(true);
|
|
try
|
|
{
|
|
await Task.Delay(Timeout.InfiniteTimeSpan, ct).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
if (ct.IsCancellationRequested)
|
|
{
|
|
RecreateCanceled.TrySetResult(true);
|
|
}
|
|
}
|
|
}
|
|
|
|
private int m_recreateCount;
|
|
}
|
|
|
|
private sealed class FatalOnceParticipant : IReconnectParticipant
|
|
{
|
|
public FatalOnceParticipant(string id, ConfiguredEndpoint endpoint)
|
|
{
|
|
Id = id;
|
|
Endpoint = endpoint;
|
|
}
|
|
|
|
public string Id { get; }
|
|
|
|
public ConfiguredEndpoint Endpoint { get; }
|
|
|
|
public ValueTask<ParticipantReconnectResult> OnReconnectAsync(
|
|
IManagedTransportChannel channel,
|
|
int reconnectAttempt,
|
|
CancellationToken ct)
|
|
{
|
|
if (reconnectAttempt < 0)
|
|
{
|
|
return new ValueTask<ParticipantReconnectResult>(
|
|
ParticipantReconnectResult.Reactivated);
|
|
}
|
|
|
|
return new ValueTask<ParticipantReconnectResult>(
|
|
Interlocked.Exchange(ref m_fatalReturned, 1) == 0
|
|
? ParticipantReconnectResult.FatalForChannel
|
|
: ParticipantReconnectResult.Reactivated);
|
|
}
|
|
|
|
private int m_fatalReturned;
|
|
}
|
|
}
|
|
}
|