mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-18 22:55:08 +08:00
Marc Schier
4fd4535362
# Description `CreateChannelAsync` now releases transferred certificate/chain handles when disposal or argument validation fails before transport creation. Managed entries use the existing inherited channel-creation method directly and share entry-retained, idempotently disposed handles with the transport, without adding another reference for non-adopting transports. The ownership contract is documented on the existing method; no internal redeclaration or forwarding implementation is needed. New and changed ownership APIs have multiline XML summaries and parameter documentation. Fourteen regression cases use generated certificates persisted in an isolated directory store and resolved through the real `CertificateManager.CertificateProvider`. They sign and verify SHA-256 digests to prove private-key usability through failed open/retry, publication/rotation, forward/reverse connection, and shutdown, and probe each certificate core's final retirement independently of global counters. Both adopting and non-adopting transports and optional chains are covered. Temporary stores are cleaned up. ## Related Issues Related to #4511: user-approved independent Wave A group 2 extraction. This PR does not close the overall tracking issue or include the other extraction groups. _If this is a relatively large or complex change, a design must have been discussed in the related tracking issue and signed off (which becomes the Architectural Decision Record (ADR))._ ## Checklist _Put an `x` in the boxes that apply. You can complete these step by step after opening the PR._ - [ ] 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. - [ ] I have added tests that prove my fix is effective or that my feature works and increased code coverage. - [x] 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. Tests were added and verified as described above; the coverage checklist remains unchecked because coverage was not measured.
1409 lines
61 KiB
C#
1409 lines
61 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/
|
|
* ======================================================================*/
|
|
|
|
using System;
|
|
using System.Collections.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.Diagnostics.Metrics;
|
|
using System.IO;
|
|
using System.Linq;
|
|
using System.Security.Cryptography;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Microsoft.Extensions.Logging;
|
|
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];
|
|
|
|
/// <summary>
|
|
/// Verifies failures before transport creation release store-backed certificate handles and chains.
|
|
/// </summary>
|
|
/// <param name="failure">The validation failure to trigger.</param>
|
|
/// <param name="includeChain">Whether the caller transfers a certificate chain as well.</param>
|
|
[Test]
|
|
[NonParallelizable]
|
|
public async Task EarlyCreateFailureReleasesStorePrivateKeyReferencesAsync(
|
|
[Values("endpoint", "context", "disposed")] string failure,
|
|
[Values] bool includeChain)
|
|
{
|
|
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
|
|
string storePath = Path.Combine(Path.GetTempPath(), "opcua-channel-" + Guid.NewGuid().ToString("N"));
|
|
try
|
|
{
|
|
await using var provider = new CertificateManager(telemetry);
|
|
using Certificate certificate = await LoadStoredCertificateAsync(
|
|
provider, storePath, "early-failure", telemetry).ConfigureAwait(false);
|
|
var settings = new ConcurrentQueue<TransportChannelSettings>();
|
|
await using ClientChannelManager manager = CreateSut(
|
|
certificate, new TestCertificateChangeSource(), settings);
|
|
Certificate transferred = certificate.AddRef();
|
|
CertificateCollection? chain = includeChain ? [certificate] : null;
|
|
try
|
|
{
|
|
var endpoint = new ConfiguredEndpoint
|
|
{
|
|
Configuration = new EndpointConfiguration()
|
|
};
|
|
IServiceMessageContext context = ServiceMessageContext.Create(telemetry);
|
|
if (failure == "disposed")
|
|
{
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
async Task CreateAsync()
|
|
{
|
|
await manager.CreateChannelAsync(
|
|
failure == "endpoint" ? null! : endpoint,
|
|
failure == "context" ? null! : context,
|
|
transferred,
|
|
chain).ConfigureAwait(false);
|
|
}
|
|
if (failure == "disposed")
|
|
{
|
|
await Assert.ThatAsync(CreateAsync, Throws.InstanceOf<ObjectDisposedException>())
|
|
.ConfigureAwait(false);
|
|
}
|
|
else
|
|
{
|
|
await Assert.ThatAsync(CreateAsync, Throws.InstanceOf<ArgumentNullException>()
|
|
.With.Property(nameof(ArgumentNullException.ParamName)).EqualTo(failure))
|
|
.ConfigureAwait(false);
|
|
}
|
|
Assert.That(settings, Is.Empty);
|
|
AssertStoredPrivateKeyWorks(certificate);
|
|
|
|
certificate.Dispose();
|
|
await provider.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(() => certificate.AddRef().Dispose(), Throws.InstanceOf<ObjectDisposedException>(),
|
|
"No transferred handle may keep the store-loaded private key alive after validation fails.");
|
|
}
|
|
finally
|
|
{
|
|
chain?.Dispose();
|
|
transferred.Dispose();
|
|
DisposeOpenedCertificates(settings);
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
if (Directory.Exists(storePath))
|
|
{
|
|
Directory.Delete(storePath, recursive: true);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies store-loaded keys survive failed opens and rotation, then retire at manager disposal.
|
|
/// </summary>
|
|
/// <param name="transportOwnsSettings">Whether the transport disposes the supplied client handles.</param>
|
|
/// <param name="includeChain">Whether each publication includes a leaf certificate chain.</param>
|
|
/// <param name="reverse">Whether the initial opens use a reverse-connect waiting connection.</param>
|
|
[Test]
|
|
[NonParallelizable]
|
|
public async Task StorePrivateKeysSurviveOpenFailureRotationAndDisposeAsync(
|
|
[Values] bool transportOwnsSettings,
|
|
[Values] bool includeChain,
|
|
[Values] bool reverse)
|
|
{
|
|
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
|
|
string storePath = Path.Combine(Path.GetTempPath(), "opcua-channel-" + Guid.NewGuid().ToString("N"));
|
|
try
|
|
{
|
|
await using var provider = new CertificateManager(telemetry);
|
|
using Certificate original = await LoadStoredCertificateAsync(
|
|
provider, storePath, "original", telemetry).ConfigureAwait(false);
|
|
using Certificate replacement = await LoadStoredCertificateAsync(
|
|
provider, storePath, "replacement", telemetry).ConfigureAwait(false);
|
|
var settings = new ConcurrentQueue<TransportChannelSettings>();
|
|
bool failOpen = true;
|
|
var bindings = new Mock<ITransportChannelBindings>();
|
|
bindings.Setup(value => value.Create(It.IsAny<string>(), It.IsAny<ITelemetryContext>()))
|
|
.Returns(() =>
|
|
{
|
|
TransportChannelSettings? ownedSettings = null;
|
|
var transport = new Mock<IChannel>();
|
|
ValueTask Open(TransportChannelSettings opened)
|
|
{
|
|
ownedSettings = opened;
|
|
settings.Enqueue(opened);
|
|
AssertStoredPrivateKeyWorks(opened.ClientCertificate!);
|
|
return failOpen
|
|
? new ValueTask(Task.FromException(
|
|
new ServiceResultException(StatusCodes.BadCommunicationError)))
|
|
: default;
|
|
}
|
|
transport.Setup(value => value.OpenAsync(
|
|
It.IsAny<Uri>(), It.IsAny<TransportChannelSettings>(), It.IsAny<CancellationToken>()))
|
|
.Returns<Uri, TransportChannelSettings, CancellationToken>((_, opened, _) => Open(opened));
|
|
transport.Setup(value => value.OpenAsync(
|
|
It.IsAny<ITransportWaitingConnection>(),
|
|
It.IsAny<TransportChannelSettings>(), It.IsAny<CancellationToken>()))
|
|
.Returns<ITransportWaitingConnection, TransportChannelSettings, CancellationToken>(
|
|
(_, opened, _) => Open(opened));
|
|
transport.Setup(value => value.CloseAsync(It.IsAny<CancellationToken>()))
|
|
.Returns(default(ValueTask));
|
|
transport.Setup(value => value.Dispose()).Callback(() =>
|
|
{
|
|
if (transportOwnsSettings && ownedSettings != null)
|
|
{
|
|
DisposeOpenedCertificates(ownedSettings);
|
|
}
|
|
});
|
|
return transport.Object;
|
|
});
|
|
await using var manager = new ClientChannelManager(
|
|
new ApplicationConfiguration(telemetry), telemetry, bindings.Object,
|
|
new ImmediateReconnectPolicy());
|
|
var endpoint = new ConfiguredEndpoint(null, new EndpointDescription
|
|
{
|
|
EndpointUrl = "opc.tcp://localhost:4840",
|
|
SecurityMode = MessageSecurityMode.None,
|
|
SecurityPolicyUri = SecurityPolicies.None,
|
|
TransportProfileUri = Profiles.UaTcpTransport
|
|
}, new EndpointConfiguration());
|
|
var participant = new TestParticipant("store-backed", endpoint);
|
|
Task<IManagedTransportChannel> GetAsync()
|
|
{
|
|
return reverse
|
|
? manager.GetAsync(participant, new Mock<ITransportWaitingConnection>().Object).AsTask()
|
|
: manager.GetAsync(participant).AsTask();
|
|
}
|
|
try
|
|
{
|
|
manager.UpdateClientCertificate(
|
|
original.AddRef(), includeChain ? [original] : null);
|
|
ServiceResultException error = Assert.ThrowsAsync<ServiceResultException>(GetAsync)!;
|
|
Assert.That(error.StatusCode, Is.EqualTo(StatusCodes.BadCommunicationError));
|
|
AssertStoredPrivateKeyWorks(original);
|
|
failOpen = false;
|
|
using IManagedTransportChannel lease = await GetAsync().ConfigureAwait(false);
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(settings, Has.Count.EqualTo(2));
|
|
TransportChannelSettings installed = settings.Last();
|
|
Assert.That(installed.ClientCertificate, Is.Not.SameAs(original));
|
|
Assert.That(installed.ClientCertificateChain != null, Is.EqualTo(includeChain));
|
|
|
|
manager.UpdateClientCertificate(
|
|
replacement.AddRef(), includeChain ? [replacement] : null);
|
|
AssertStoredPrivateKeyWorks(installed.ClientCertificate!);
|
|
if (reverse)
|
|
{
|
|
// Reverse recovery requires another waiting connection rather than reconnecting the old one.
|
|
await lease.CloseAsync().ConfigureAwait(false);
|
|
}
|
|
else
|
|
{
|
|
await manager.ReconnectAsync(lease).ConfigureAwait(false);
|
|
}
|
|
using IManagedTransportChannel current = reverse
|
|
? await GetAsync().ConfigureAwait(false)
|
|
: lease;
|
|
Assert.That(current.State, Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(settings.Last().ClientCertificate!.Thumbprint, Is.EqualTo(replacement.Thumbprint));
|
|
AssertStoredPrivateKeyWorks(settings.Last().ClientCertificate!);
|
|
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
Assert.That(current.State, Is.EqualTo(ChannelState.Closed));
|
|
Assert.That(manager.GetChannelDiagnostics(), Is.Empty);
|
|
AssertStoredPrivateKeyWorks(original);
|
|
AssertStoredPrivateKeyWorks(replacement);
|
|
original.Dispose();
|
|
replacement.Dispose();
|
|
await provider.DisposeAsync().ConfigureAwait(false);
|
|
Assert.Multiple(() =>
|
|
{
|
|
Assert.That(() => original.AddRef().Dispose(), Throws.InstanceOf<ObjectDisposedException>(),
|
|
"The original key must retire without relying on transport settings cleanup.");
|
|
Assert.That(() => replacement.AddRef().Dispose(), Throws.InstanceOf<ObjectDisposedException>(),
|
|
"The replacement key must retire without relying on transport settings cleanup.");
|
|
});
|
|
}
|
|
finally
|
|
{
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
DisposeOpenedCertificates(settings);
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
if (Directory.Exists(storePath))
|
|
{
|
|
Directory.Delete(storePath, recursive: true);
|
|
}
|
|
}
|
|
}
|
|
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task QueuedRotationRetainsProducerCertificateMaterialAsync(bool sendChain)
|
|
{
|
|
using Certificate original = s_factory.CreateCertificate("CN=queued-original").CreateForRSA();
|
|
using Certificate first = s_factory.CreateCertificate("CN=queued-first").CreateForRSA();
|
|
var changes = new TestCertificateChangeSource();
|
|
var certificateManager = new Mock<ICertificateManager>();
|
|
certificateManager.SetupGet(value => value.CertificateChanges).Returns(changes);
|
|
var configuration = new ApplicationConfiguration(NUnitTelemetryContext.Create())
|
|
{
|
|
CertificateManager = certificateManager.Object
|
|
};
|
|
configuration.SecurityConfiguration.ApplicationCertificate = new CertificateIdentifier
|
|
{
|
|
CertificateType = ObjectTypeIds.RsaSha256ApplicationCertificateType,
|
|
Thumbprint = original.Thumbprint
|
|
};
|
|
configuration.SecurityConfiguration.SendCertificateChain = sendChain;
|
|
var installed = new ConcurrentQueue<(Certificate Certificate, CertificateCollection? Chain)>();
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var started = new TaskCompletionSource<Task>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
using var release = new ManualResetEventSlim();
|
|
var host = new RotationHost(
|
|
configuration,
|
|
task =>
|
|
{
|
|
if (task != null)
|
|
{
|
|
started.TrySetResult(task);
|
|
}
|
|
},
|
|
(certificate, chain) =>
|
|
{
|
|
if (installed.IsEmpty)
|
|
{
|
|
entered.TrySetResult(true);
|
|
if (!release.Wait(TimeSpan.FromSeconds(5)))
|
|
{
|
|
throw new TimeoutException("The first certificate installation was not released.");
|
|
}
|
|
}
|
|
installed.Enqueue((certificate, chain));
|
|
});
|
|
using var rotation = new ClientChannelManagerCertRotation(host);
|
|
rotation.WireCertificateRotation();
|
|
Task drain = Task.CompletedTask;
|
|
try
|
|
{
|
|
changes.Raise(new CertificateChangeEvent(
|
|
CertificateChangeKind.ApplicationCertificateUpdated, TrustListIdentifier.Peers,
|
|
ObjectTypeIds.RsaSha256ApplicationCertificateType, original, first, null));
|
|
drain = await started.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
string thumbprint;
|
|
using (Certificate next = s_factory.CreateCertificate("CN=queued-next").CreateForRSA())
|
|
using (CertificateCollection chain = [next])
|
|
{
|
|
thumbprint = next.Thumbprint;
|
|
changes.Raise(new CertificateChangeEvent(
|
|
CertificateChangeKind.ApplicationCertificateUpdated, TrustListIdentifier.Peers,
|
|
ObjectTypeIds.RsaSha256ApplicationCertificateType, original, next, chain));
|
|
}
|
|
release.Set();
|
|
await drain.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
Assert.That(installed, Has.Count.EqualTo(2));
|
|
Assert.That(installed.Last().Certificate.Thumbprint, Is.EqualTo(thumbprint));
|
|
Assert.That(installed.Last().Certificate.HasPrivateKey, Is.True);
|
|
Assert.That(installed.Last().Chain != null, Is.EqualTo(sendChain));
|
|
}
|
|
finally
|
|
{
|
|
release.Set();
|
|
await drain.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
while (installed.TryDequeue(out (Certificate Certificate, CertificateCollection? Chain) material))
|
|
{
|
|
material.Certificate.Dispose();
|
|
material.Chain?.Dispose();
|
|
}
|
|
}
|
|
}
|
|
|
|
[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);
|
|
|
|
DisposeOpenedCertificates(openSettings);
|
|
}
|
|
}
|
|
|
|
[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);
|
|
|
|
DisposeOpenedCertificates(openSettings);
|
|
}
|
|
}
|
|
|
|
[TestCase(false)]
|
|
[TestCase(true)]
|
|
public async Task ReconnectUsesLatestCertificateForEachEntryTypeAsync(bool reusableTransport)
|
|
{
|
|
using Certificate oldRsa = s_factory.CreateCertificate("CN=old-rsa").CreateForRSA();
|
|
using Certificate newRsa = s_factory.CreateCertificate("CN=new-rsa").CreateForRSA();
|
|
using Certificate oldEcc = s_factory.CreateCertificate("CN=old-ecc")
|
|
.SetECCurve(ECCurve.NamedCurves.nistP256).CreateForECDsa();
|
|
using Certificate newEcc = s_factory.CreateCertificate("CN=new-ecc")
|
|
.SetECCurve(ECCurve.NamedCurves.nistP256).CreateForECDsa();
|
|
using Certificate server = s_factory.CreateCertificate("CN=server").CreateForRSA();
|
|
var changes = new TestCertificateChangeSource();
|
|
var settings = new ConcurrentQueue<TransportChannelSettings>();
|
|
await using ClientChannelManager manager = CreateSut(
|
|
oldRsa, changes, settings, reusableTransport: reusableTransport);
|
|
ConfiguredEndpoint rsaEndpoint = GetTestEndpoint(server);
|
|
ConfiguredEndpoint eccEndpoint = GetTestEndpoint(server);
|
|
eccEndpoint.Description.EndpointUrl = "opc.tcp://localhost:4841";
|
|
eccEndpoint.Description.SecurityPolicyUri = SecurityPolicies.ECC_nistP256;
|
|
try
|
|
{
|
|
manager.UpdateClientCertificate(oldRsa.AddRef(), null);
|
|
using IManagedTransportChannel rsa = await manager.GetAsync(new TestParticipant("rsa", rsaEndpoint))
|
|
.ConfigureAwait(false);
|
|
manager.UpdateClientCertificate(oldEcc.AddRef(), null);
|
|
using IManagedTransportChannel ecc = await manager.GetAsync(new TestParticipant("ecc", eccEndpoint))
|
|
.ConfigureAwait(false);
|
|
|
|
await manager.ReconnectAllAsync().ConfigureAwait(false);
|
|
Assert.That(settings.Last(value => value.Description?.EndpointUrl == rsa.Key.EndpointUrl)
|
|
.ClientCertificate!.Thumbprint, Is.EqualTo(oldRsa.Thumbprint));
|
|
|
|
manager.UpdateClientCertificate(newRsa.AddRef(), null);
|
|
manager.UpdateClientCertificate(newEcc.AddRef(), null);
|
|
await manager.ReconnectAllAsync().ConfigureAwait(false);
|
|
|
|
Assert.Multiple(() =>
|
|
{
|
|
Assert.That(settings.Last(value => value.Description?.EndpointUrl == rsa.Key.EndpointUrl)
|
|
.ClientCertificate!.Thumbprint, Is.EqualTo(newRsa.Thumbprint));
|
|
Assert.That(settings.Last(value => value.Description?.EndpointUrl == ecc.Key.EndpointUrl)
|
|
.ClientCertificate!.Thumbprint, Is.EqualTo(newEcc.Thumbprint));
|
|
Assert.That(manager.GetChannelDiagnostics().Select(value => value.Refcount), Is.All.EqualTo(1));
|
|
Assert.That(rsa.State, Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(ecc.State, Is.EqualTo(ChannelState.Ready));
|
|
});
|
|
int opened = settings.Count;
|
|
await manager.ReconnectAllAsync().ConfigureAwait(false);
|
|
Assert.That(settings, Has.Count.EqualTo(reusableTransport ? opened : opened + 2));
|
|
}
|
|
finally
|
|
{
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
DisposeOpenedCertificates(settings);
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReconnectCompletionWaitsForCycleCleanupAsync()
|
|
{
|
|
using Certificate client = s_factory.CreateCertificate("CN=cleanup-client").CreateForRSA();
|
|
using Certificate server = s_factory.CreateCertificate("CN=cleanup-server").CreateForRSA();
|
|
var settings = new ConcurrentQueue<TransportChannelSettings>();
|
|
var changes = new TestCertificateChangeSource();
|
|
await using ClientChannelManager manager = CreateSut(client, changes, settings);
|
|
ConfiguredEndpoint endpoint = GetTestEndpoint(server);
|
|
endpoint.Description.EndpointUrl = "opc.tcp://localhost:4842/reconnect-cleanup";
|
|
var participant = new BlockingRecreateParticipant("cleanup", endpoint);
|
|
var completedDuringCleanup = new TaskCompletionSource<bool>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
Task<bool>? reconnect = null;
|
|
using var listener = new MeterListener
|
|
{
|
|
InstrumentPublished = (instrument, meterListener) =>
|
|
{
|
|
if (instrument.Name == "opc.ua.channel.reconnect.duration")
|
|
{
|
|
meterListener.EnableMeasurementEvents(instrument);
|
|
}
|
|
}
|
|
};
|
|
listener.SetMeasurementEventCallback<double>((_, _, tags, _) =>
|
|
{
|
|
foreach (KeyValuePair<string, object?> tag in tags)
|
|
{
|
|
if (tag.Key == "endpoint" &&
|
|
tag.Value is string endpointUrl &&
|
|
endpointUrl == endpoint.Description.EndpointUrl)
|
|
{
|
|
completedDuringCleanup.TrySetResult(reconnect!.IsCompleted);
|
|
}
|
|
}
|
|
});
|
|
listener.Start();
|
|
ManagedTransportChannelLease? lease = null;
|
|
try
|
|
{
|
|
manager.UpdateClientCertificate(client.AddRef(), null);
|
|
lease = (ManagedTransportChannelLease)await manager.GetAsync(participant).ConfigureAwait(false);
|
|
reconnect = lease.Entry.RequestReconnectAsync(default);
|
|
await participant.RecreateStarted.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
participant.ReleaseRecreate();
|
|
|
|
Assert.That(await reconnect.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false), Is.True);
|
|
Assert.That(
|
|
await completedDuringCleanup.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false),
|
|
Is.False,
|
|
"Reconnect waiters must not resume before the completed cycle releases its coalescer.");
|
|
Assert.That(lease.State, Is.EqualTo(ChannelState.Ready));
|
|
Assert.That(lease.Entry.RefCount, Is.EqualTo(1));
|
|
}
|
|
finally
|
|
{
|
|
participant.ReleaseRecreate();
|
|
if (lease != null)
|
|
{
|
|
await lease.CloseAsync().ConfigureAwait(false);
|
|
}
|
|
await manager.DisposeAsync().ConfigureAwait(false);
|
|
DisposeOpenedCertificates(settings);
|
|
}
|
|
}
|
|
|
|
[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);
|
|
DisposeOpenedCertificates(openSettings);
|
|
}
|
|
}
|
|
|
|
[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);
|
|
DisposeOpenedCertificates(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);
|
|
DisposeOpenedCertificates(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);
|
|
DisposeOpenedCertificates(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);
|
|
DisposeOpenedCertificates(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 sealed class RotationHost(
|
|
ApplicationConfiguration configuration,
|
|
Action<Task?> stateChanged,
|
|
Action<Certificate, CertificateCollection?> install) : IChannelCertRotationHost
|
|
{
|
|
public ApplicationConfiguration Configuration => configuration;
|
|
|
|
public ILogger? Logger => null;
|
|
|
|
public bool IsDisposed => false;
|
|
|
|
public ChannelEntry[] SnapshotEntries()
|
|
{
|
|
return [];
|
|
}
|
|
|
|
public void ReplaceClientCertificate(Certificate? certificate, CertificateCollection? chain)
|
|
{
|
|
install(certificate ?? throw new ArgumentNullException(nameof(certificate)), chain);
|
|
}
|
|
|
|
public void SetCertificateRotationTask(Task? task)
|
|
{
|
|
stateChanged(task);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Persists a generated certificate and resolves its private key through the real store-backed provider.
|
|
/// </summary>
|
|
/// <param name="provider">The certificate manager whose provider resolves the stored private key.</param>
|
|
/// <param name="storePath">The isolated directory certificate store.</param>
|
|
/// <param name="name">The subject distinguishing this certificate.</param>
|
|
/// <param name="telemetry">The telemetry passed to the certificate store.</param>
|
|
/// <returns>A caller-owned store-loaded certificate with its private key.</returns>
|
|
/// <exception cref="InvalidOperationException">The stored private key could not be resolved.</exception>
|
|
private static async Task<Certificate> LoadStoredCertificateAsync(
|
|
CertificateManager provider,
|
|
string storePath,
|
|
string name,
|
|
ITelemetryContext telemetry)
|
|
{
|
|
using Certificate created = s_factory.CreateCertificate($"CN=channel-{name}").CreateForRSA();
|
|
await created.AddToStoreAsync(
|
|
CertificateStoreType.Directory, storePath, password: null, telemetry).ConfigureAwait(false);
|
|
return await provider.CertificateProvider.GetPrivateKeyCertificateAsync(new CertificateIdentifier
|
|
{
|
|
StoreType = CertificateStoreType.Directory,
|
|
StorePath = storePath,
|
|
Thumbprint = created.Thumbprint
|
|
}).ConfigureAwait(false) ??
|
|
throw new InvalidOperationException("The stored private key was not resolved.");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Signs and verifies a SHA-256 digest using the store-loaded RSA key.
|
|
/// </summary>
|
|
/// <param name="certificate">The certificate whose private key must remain usable.</param>
|
|
/// <exception cref="InvalidOperationException">The certificate has no RSA private or public key.</exception>
|
|
private static void AssertStoredPrivateKeyWorks(Certificate certificate)
|
|
{
|
|
using RSA key = certificate.GetRSAPrivateKey() ?? throw new InvalidOperationException("No private key.");
|
|
using RSA verifier = certificate.GetRSAPublicKey() ??
|
|
throw new InvalidOperationException("No public key.");
|
|
byte[] hash = new byte[32];
|
|
byte[] signature = key.SignHash(hash, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
|
|
Assert.That(
|
|
verifier.VerifyHash(hash, signature, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1), Is.True);
|
|
}
|
|
|
|
private static ClientChannelManager CreateSut(
|
|
Certificate applicationCertificate,
|
|
TestCertificateChangeSource changes,
|
|
ConcurrentQueue<TransportChannelSettings> openSettings,
|
|
IChannelReconnectPolicy? reconnectPolicy = null,
|
|
bool reusableTransport = false)
|
|
{
|
|
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(reusableTransport
|
|
? TransportChannelFeatures.Reconnect
|
|
: 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);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Releases the certificate handles a real channel would own.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// A production channel takes over everything on the settings and
|
|
/// releases it when it closes: the server certificate it parses out of
|
|
/// the description, and the client certificate and chain the manager
|
|
/// hands it a reference of its own on. The mock channel captures the
|
|
/// settings and does neither, so the test stands in for it.
|
|
/// </remarks>
|
|
private static void DisposeOpenedCertificates(
|
|
ConcurrentQueue<TransportChannelSettings> openSettings)
|
|
{
|
|
while (openSettings.TryDequeue(out TransportChannelSettings? opened))
|
|
{
|
|
DisposeOpenedCertificates(opened);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Releases one transport's certificate handles in dependency order.
|
|
/// </summary>
|
|
/// <param name="opened">The settings whose handles are no longer needed.</param>
|
|
private static void DisposeOpenedCertificates(TransportChannelSettings opened)
|
|
{
|
|
opened.ServerCertificate?.Dispose();
|
|
opened.ClientCertificateChain?.Dispose();
|
|
opened.ClientCertificate?.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);
|
|
|
|
/// <inheritdoc/>
|
|
public IRetryBudget? CreateReconnectBudget(TimeProvider timeProvider)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
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);
|
|
|
|
/// <inheritdoc/>
|
|
public IRetryBudget? CreateReconnectBudget(TimeProvider timeProvider)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
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];
|
|
|
|
/// <inheritdoc/>
|
|
public IRetryBudget? CreateReconnectBudget(TimeProvider timeProvider)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
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);
|
|
|
|
/// <inheritdoc/>
|
|
public IRetryBudget? CreateReconnectBudget(TimeProvider timeProvider)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
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; }
|
|
|
|
/// <inheritdoc/>
|
|
public IRetryBudget? CreateReconnectBudget(TimeProvider timeProvider)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
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;
|
|
}
|
|
}
|
|
}
|