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

test: verify live history authority and secure peer reactivation

Browse Source
This commit is contained in:
Joseph Doherty
2026-11-10 11:53:34 -05:00
parent da50123a2c
commit 1840b5ccf6
2 changed files with 175 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2HistoryAuthorityWireTests.cs
+156
View File
@@ -0,0 +1,156 @@
using Opc.Ua;
using Opc.Ua.Client;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
using CoreReadResult = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// <summary>Uses activated sessions and the production continuation store; denied resumes never reach a backend.</summary>
public sealed class Sdk2HistoryAuthorityWireTests
{
private const string ValuePath = RawDeviceFolder + "/History";
private static readonly DateTime Epoch = new(2026, 10, 1, 0, 0, 0, DateTimeKind.Utc);
public static TheoryData<string, string> Cases => new(
from kind in new[] { "raw", "modified", "events" }
from change in new[] { "alias", "replace", "rebuild", "grant", "key", "expiry" }
select (kind, change));
[Theory]
[MemberData(nameof(Cases))]
public async Task A_saved_cursor_requires_its_live_node_and_current_authority(string kind, string change)
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await StartAsync("HistoryAuthority", encrypted: true, ct);
var source = new CountingSource();
var clock = new Clock();
var key = Key();
key.ExpiresUtc = clock.GetUtcNow().UtcDateTime.AddMinutes(10);
var grant = ClusterGrant(KeyPrincipal, "history-allow", NodePermissions.ReadOnly);
var state = rig.Enforce(true, [key], grant);
rig.Nm.Authorizer = new DataPlaneAuthorizer(state, new TriePermissionEvaluator(state.Tries), true,
NullLogger.Instance, clock);
rig.Nm.HistorianDataSource = source;
Materialise(rig);
using var session = await rig.ConnectAsync(anonymous: false, ct, encrypted: true);
var first = await PageAsync(session, kind, alias: false, default, ct);
first.StatusCode.ShouldBe(StatusCodes.Good);
first.ContinuationPoint.IsEmpty.ShouldBeFalse("the production session store must own a real next page");
source.Calls.ShouldBeGreaterThan(0);
switch (change)
{
case "replace":
if (kind == "events") rig.Nm.RemoveEquipmentSubtree(RawDeviceFolder, AddressSpaceRealm.Raw).ShouldBeTrue();
else rig.Nm.RemoveVariableNode(ValuePath, AddressSpaceRealm.Raw).ShouldBeTrue();
Materialise(rig);
break;
case "rebuild":
rig.Nm.RebuildAddressSpace(); Materialise(rig); rig.Nm.CompleteAddressSpaceApply();
break;
case "grant": Apply(state, key, []); break;
case "key":
var disabled = Key(); disabled.Enabled = false; Apply(state, disabled, [grant]); break;
case "expiry": clock.Advance(TimeSpan.FromMinutes(20)); break;
}
var before = source.Calls;
var denied = await PageAsync(session, kind, change == "alias", first.ContinuationPoint, ct);
denied.StatusCode.ShouldBe(change is "grant" or "key" or "expiry"
? StatusCodes.BadUserAccessDenied : StatusCodes.BadContinuationPointInvalid);
denied.HistoryData.IsNull.ShouldBeTrue("a denied resume must not disclose saved data");
denied.ContinuationPoint.IsEmpty.ShouldBeTrue();
source.Calls.ShouldBe(before, "permission and exact-node binding must be checked before backend dispatch");
// A fresh authorized request remains usable: this is not an always-deny fixture.
clock.Reset(); Apply(state, key, [grant]);
(await PageAsync(session, kind, alias: false, default, ct)).StatusCode.ShouldBe(StatusCodes.Good);
source.Calls.ShouldBeGreaterThan(before);
await session.CloseAsync(ct);
}
private static void Apply(DataPlaneAuthorityState state, ApiKey key, NodeAcl[] grants) =>
state.Apply(ServerEventRig.Cluster, Guid.NewGuid().ToString("N"), [key], grants,
new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>());
private static void Materialise(ServerEventRig rig)
{
rig.MaterialiseNative();
rig.Sink.EnsureVariable(ValuePath, RawDeviceFolder, "History", "Double", false, AddressSpaceRealm.Raw, "same-backend-tag");
rig.Sink.EnsureVariable(ValuePath, EqA, "HistoryAlias", "Double", false, AddressSpaceRealm.Uns, "same-backend-tag");
}
private static async Task<Opc.Ua.HistoryReadResult> PageAsync(ISession session, string kind, bool alias,
ByteString point, CancellationToken ct)
{
var realm = alias ? V3NodeIds.UnsNamespaceUri : V3NodeIds.RawNamespaceUri;
var path = kind == "events" ? alias ? EqA : RawDeviceFolder : ValuePath;
var node = new NodeId(path, (ushort)session.NamespaceUris.GetIndex(realm));
HistoryReadDetails details = kind == "events" ? new ReadEventDetails
{
StartTime = Epoch, EndTime = Epoch.AddMinutes(1), NumValuesPerNode = 2,
Filter = new EventFilter { SelectClauses =
[new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, QualifiedName.From(BrowseNames.EventId))] },
} : new ReadRawModifiedDetails
{ StartTime = Epoch, EndTime = Epoch.AddMinutes(1), NumValuesPerNode = 2, IsReadModified = kind == "modified" };
var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, false,
[new HistoryReadValueId { NodeId = node, ContinuationPoint = point }], ct);
return response.Results.ToList().ShouldHaveSingleItem();
}
private sealed class Clock : TimeProvider
{
private readonly DateTimeOffset _initial = DateTimeOffset.UtcNow;
private TimeSpan _offset;
public override DateTimeOffset GetUtcNow() => _initial + _offset;
public void Advance(TimeSpan amount) => _offset += amount;
public void Reset() => _offset = TimeSpan.Zero;
}
private sealed class CountingSource : IHistorianDataSource
{
private int _calls;
public int Calls => Volatile.Read(ref _calls);
private static DataValueSnapshot Value(int i) => new((double)i, OpcUaStatus.Good, Epoch.AddSeconds(i), Epoch.AddSeconds(i));
public Task<CoreReadResult> ReadRawAsync(string reference, DateTime start, DateTime end, uint max, CancellationToken ct)
{
Interlocked.Increment(ref _calls);
var rows = Enumerable.Range(0, 25).Select(Value).Where(x => x.SourceTimestampUtc >= start && x.SourceTimestampUtc < end).ToList();
var count = max == 0 ? rows.Count : checked((int)max);
return Task.FromResult(new CoreReadResult(rows.Take(count).ToList(), rows.Count > count ? [1] : null));
}
public Task<HistoryModifiedReadResult> ReadModifiedAsync(string reference, DateTime start, DateTime end,
uint max, HistoryModifiedCursor? resume, CancellationToken ct)
{
Interlocked.Increment(ref _calls);
var rows = Enumerable.Range(0, 25).Where(i => Epoch.AddSeconds(i) >= start && Epoch.AddSeconds(i) < end
&& (resume is null || i > resume.ModifiedId))
.Select(i => new HistoryModifiedValue(Value(i), Epoch.AddMinutes(2), HistoryModificationType.Replace,
"fixture", i, Epoch.AddSeconds(i))).ToList();
var count = max == 0 ? rows.Count : checked((int)max);
return Task.FromResult(new HistoryModifiedReadResult(rows.Take(count).ToList(), rows.Count > count));
}
public Task<HistoricalEventsResult> ReadEventsAsync(string? source, DateTime start, DateTime end, int max, CancellationToken ct)
{
Interlocked.Increment(ref _calls);
var rows = Enumerable.Range(0, 25).Select(i => new HistoricalEvent($"e{i:00}", "src", Epoch.AddSeconds(i), Epoch, "m", 500))
.Where(x => x.EventTimeUtc >= start && x.EventTimeUtc < end).ToList();
var count = max <= 0 ? rows.Count : max;
return Task.FromResult(new HistoricalEventsResult(rows.Take(count).ToList(), rows.Count > count ? [1] : null));
}
public Task<CoreReadResult> ReadProcessedAsync(string r, DateTime s, DateTime e, TimeSpan interval, HistoryAggregateType a, CancellationToken ct) =>
NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, interval, a, ct);
public Task<CoreReadResult> ReadAtTimeAsync(string r, IReadOnlyList<DateTime> times, CancellationToken ct) =>
NullHistorianDataSource.Instance.ReadAtTimeAsync(r, times, ct);
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
public void Dispose() { }
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2TrustedChannelTests.cs
+19
View File
@@ -1,4 +1,6 @@
using System.Text;
using System.Net;
using System.Collections.Concurrent;
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Client;
@@ -29,6 +31,16 @@ public sealed class Sdk2TrustedChannelTests
(await session.ReadValueAsync(rig.Node(session), ct)).WrappedValue.GetInt32().ShouldBe(7);
(await rig.WriteAsync(session, 42, ct)).ShouldBe(StatusCodes.Good);
rig.Writes.ShouldBe(1, "a narrow grant with no application-wide roles must permit the write");
var observedPeer = rig.Peers.ShouldHaveSingleItem();
observedPeer.ShouldNotBeNull();
IPAddress.IsLoopback(observedPeer.Address).ShouldBeTrue();
observedPeer.Port.ShouldBe(0, "the public SDK evidence supplies an IP address, not an invented client port");
var originalSessionId = session.SessionId;
await session.ReconnectAsync(null, null, ct);
session.SessionId.ShouldBe(originalSessionId, "the test must reactivate the existing session on a new channel");
rig.Peers.Count.ShouldBe(2);
rig.Peers.Last().ShouldNotBeNull().Address.ShouldBe(observedPeer.Address);
(await session.ReadValueAsync(rig.Node(session), ct)).StatusCode.ShouldBe(StatusCodes.Good);
var identity = rig.Server.CurrentInstance.SessionManager.GetSessions().ShouldHaveSingleItem().EffectiveIdentity
.ShouldBeOfType<RoleCarryingUserIdentity>();
identity.Principal.ShouldBe(TrustedRig.Principal);
@@ -80,6 +92,7 @@ public sealed class Sdk2TrustedChannelTests
private int _port;
private int _writes;
private int _authCalls;
internal ConcurrentQueue<IPEndPoint?> Peers { get; } = new();
internal OtOpcUaSdkServer Server { get; private set; } = null!;
internal int Writes => Volatile.Read(ref _writes);
internal int AuthCalls => Volatile.Read(ref _authCalls);
@@ -204,6 +217,12 @@ public sealed class Sdk2TrustedChannelTests
new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>
{ [new(NodeAclScopeKind.RawDevice, "device")] = [new([NodeAclRealms.Raw, "Plant", "Device"])] });
public Task<OpcUaUserAuthResult> AuthenticateUserNameAsync(string username, string password, IPEndPoint? peer, CancellationToken ct)
{
Peers.Enqueue(peer is null ? null : new IPEndPoint(peer.Address, peer.Port));
return AuthenticateUserNameAsync(username, password, ct);
}
public Task<OpcUaUserAuthResult> AuthenticateUserNameAsync(string username, string password, CancellationToken ct)
{
Interlocked.Increment(ref _authCalls);