diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2HistoryAuthorityWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2HistoryAuthorityWireTests.cs new file mode 100644 index 00000000..d5008082 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2HistoryAuthorityWireTests.cs @@ -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; + +/// Uses activated sessions and the production continuation store; denied resumes never reach a backend. +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 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>()); + + 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 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 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 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 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 ReadProcessedAsync(string r, DateTime s, DateTime e, TimeSpan interval, HistoryAggregateType a, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, interval, a, ct); + public Task ReadAtTimeAsync(string r, IReadOnlyList times, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadAtTimeAsync(r, times, ct); + public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); + public void Dispose() { } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2TrustedChannelTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2TrustedChannelTests.cs index 24ccfc1a..f2695e81 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2TrustedChannelTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2TrustedChannelTests.cs @@ -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(); identity.Principal.ShouldBe(TrustedRig.Principal); @@ -80,6 +92,7 @@ public sealed class Sdk2TrustedChannelTests private int _port; private int _writes; private int _authCalls; + internal ConcurrentQueue 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> { [new(NodeAclScopeKind.RawDevice, "device")] = [new([NodeAclRealms.Raw, "Plant", "Device"])] }); + public Task 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 AuthenticateUserNameAsync(string username, string password, CancellationToken ct) { Interlocked.Increment(ref _authCalls);