Merge #587/#588: one condition event per method-driven transition (native: Ack/AddComment only); model-change + audit events gated per session on i=2253 when ACL is on
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
+5
-4
@@ -650,11 +650,12 @@ public sealed class AclEventSubscriptionEnforcementTests
|
||||
/// including a condition rebuilt by a redeploy while the item was live.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <c>ConditionState.EventsMonitored()</c> gates whether the SDK reports a condition's method-driven
|
||||
/// state changes at all. The SDK base raised it by attaching the item to every root notifier; #586
|
||||
/// <c>ConditionState.EventsMonitored()</c> gates whether the SDK reports a condition's method audit
|
||||
/// events. The SDK base raised it by attaching the item to every root notifier; #586
|
||||
/// attaches to nothing, so the first cut left it false on a server whose only subscribers are
|
||||
/// Server-object items — the machine-consumer default. Asserted server-side because the consequence on
|
||||
/// the wire depends on a separate, still-open reporting gap.
|
||||
/// Server-object items — the machine-consumer default. Asserted server-side: method-driven state-change
|
||||
/// events no longer depend on it (Gitea #587, <c>ConditionMethodEventTests</c>), and audit events only
|
||||
/// leave the server while it is auditing.
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public async Task A_server_object_item_alone_marks_conditions_monitored_until_it_is_deleted()
|
||||
|
||||
+201
@@ -0,0 +1,201 @@
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #588 over the wire — with <c>Security:Acl:Enabled</c> on, the server-wide events this node manager
|
||||
/// raises (<c>GeneralModelChangeEvent</c>s, <c>AuditWriteUpdateEvent</c>s) reach a Server-object
|
||||
/// (<c>i=2253</c>) item only when its session may see the node the event names.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// They used to go straight to <c>Server.ReportEvent</c> and the core node manager's Server object, which
|
||||
/// queues to every all-events item with no ACL decision — so a session with no grant learned, from a
|
||||
/// model-change event, the RawPath of every node added, deleted or retyped under it.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The withheld session is granted <see cref="ServerEventRig.EqA"/> only, so the absence of the raw
|
||||
/// event on its item is paired with a positive control <b>on the same item</b>: the sentinel event about
|
||||
/// <see cref="ServerEventRig.EqA"/> is reported after the withheld one, through the same path, and must
|
||||
/// arrive.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class AclServerObjectModelChangeTests
|
||||
{
|
||||
private const string RawVariable = "Plant/Modbus/dev1/Speed";
|
||||
|
||||
private const NodePermissions See = NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe;
|
||||
|
||||
/// <summary>
|
||||
/// A NodeAdded / NodeDeleted / DataTypeChanged model-change event naming a raw node reaches the
|
||||
/// cluster-granted session and not the session granted only an equipment folder — which still receives
|
||||
/// the one naming its own folder.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData("NodeAdded")]
|
||||
[InlineData("NodeDeleted")]
|
||||
[InlineData("DataTypeChanged")]
|
||||
public async Task A_model_change_naming_a_node_the_session_cannot_browse_is_withheld(string verb)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync($"AclModelChange{verb}", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(true, [Key()], ClusterGrant(KeyPrincipal, "acl-key", See), EquipmentGrant(EqA, See));
|
||||
rig.EnsureEquipment();
|
||||
rig.Sink.EnsureFolder(RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
|
||||
rig.Sink.EnsureVariable(RawVariable, RawDeviceFolder, "Speed", "Int32", writable: false, AddressSpaceRealm.Raw);
|
||||
|
||||
var (granted, grantedEvents, grantedItem, grantedSubscription) = await ServerObjectAsync(rig, anonymous: false, ct);
|
||||
var (narrow, narrowEvents, narrowItem, narrowSubscription) = await ServerObjectAsync(rig, anonymous: true, ct);
|
||||
|
||||
switch (verb)
|
||||
{
|
||||
case "NodeAdded": rig.Sink.RaiseNodesAddedModelChange(RawDeviceFolder, AddressSpaceRealm.Raw); break;
|
||||
case "NodeDeleted": rig.Sink.RemoveVariableNode(RawVariable, AddressSpaceRealm.Raw).ShouldBeTrue(); break;
|
||||
default:
|
||||
rig.Sink.UpdateTagAttributes(RawVariable, false, null, "Double", false, null, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
break;
|
||||
}
|
||||
|
||||
rig.Sink.RaiseNodesAddedModelChange(EqA, AddressSpaceRealm.Uns);
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() =>
|
||||
grantedEvents.Of(grantedItem).Count(e => NamesEqA(e)) == 1
|
||||
&& narrowEvents.Of(narrowItem).Any(NamesEqA));
|
||||
|
||||
var rawNames = verb == "NodeAdded" ? RawDeviceFolder : RawVariable;
|
||||
grantedEvents.Of(grantedItem).Count(e => Names(e, rawNames)).ShouldBe(1, "the granted control receives the raw change");
|
||||
narrowEvents.Of(narrowItem).Where(e => Names(e, rawNames)).ShouldBeEmpty(
|
||||
"a session that cannot browse the node must not learn its name from a model-change event");
|
||||
narrowEvents.Of(narrowItem).Count.ShouldBe(1, "only the change to its own equipment folder");
|
||||
|
||||
await narrowSubscription.DeleteAsync(true, ct);
|
||||
await grantedSubscription.DeleteAsync(true, ct);
|
||||
narrow.Dispose();
|
||||
granted.Dispose();
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// With enforcement off nothing changes: a session holding no grant at all still receives the raw
|
||||
/// model-change event through the Server object, exactly once.
|
||||
/// </summary>
|
||||
/// <remarks>Without this the test above also passes on a build that withholds every model change from everyone.</remarks>
|
||||
[Fact]
|
||||
public async Task With_acl_enforcement_off_every_server_object_item_receives_the_model_change()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync("AclModelChangeOff", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(false, [Key()], ClusterGrant(KeyPrincipal, "acl-key", See));
|
||||
rig.Sink.EnsureFolder(RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
|
||||
|
||||
var (session, events, item, subscription) = await ServerObjectAsync(rig, anonymous: true, ct);
|
||||
|
||||
rig.Sink.RaiseNodesAddedModelChange(RawDeviceFolder, AddressSpaceRealm.Raw);
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Of(item).Any(e => Names(e, RawDeviceFolder)));
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(300), ct);
|
||||
events.Of(item).Count(e => Names(e, RawDeviceFolder)).ShouldBe(1);
|
||||
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
session.Dispose();
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// With auditing on over an encrypted channel, a rejected device write's <c>AuditWriteUpdateEvent</c> —
|
||||
/// which carries the written value and the node's prior one — reaches a session granted <c>Read</c> on
|
||||
/// the node and not one granted only an equipment folder.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Auditing is off by default (<c>ServerConfiguration.AuditingEnabled</c>, which this repo never sets),
|
||||
/// and audit events only ever leave over SignAndEncrypt — so both sessions are encrypted here, or the
|
||||
/// withheld one would be denied by the channel rather than by the ACL and the test would prove nothing.
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public async Task An_audit_write_event_reaches_only_sessions_that_may_read_the_node()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync("AclAuditWrite", encrypted: true, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(
|
||||
true, [Key()],
|
||||
ClusterGrant(KeyPrincipal, "acl-key", See | NodePermissions.WriteOperate), EquipmentGrant(EqA, See));
|
||||
rig.SetAuditing(true);
|
||||
rig.EnsureEquipment();
|
||||
rig.Sink.EnsureFolder(RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
|
||||
rig.Sink.EnsureVariable(RawVariable, RawDeviceFolder, "Speed", "Int32", writable: true, AddressSpaceRealm.Raw);
|
||||
rig.Nm.NodeWriteGateway = new RejectingGateway();
|
||||
|
||||
var (granted, grantedEvents, grantedItem, grantedSubscription) =
|
||||
await ServerObjectAsync(rig, anonymous: false, ct, encrypted: true);
|
||||
var (narrow, narrowEvents, narrowItem, narrowSubscription) =
|
||||
await ServerObjectAsync(rig, anonymous: true, ct, encrypted: true);
|
||||
|
||||
var variable = new NodeId(RawVariable, (ushort)granted.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri));
|
||||
var write = await granted.WriteAsync(
|
||||
null,
|
||||
[new WriteValue { NodeId = variable, AttributeId = Attributes.Value, Value = new DataValue(new Variant(42)) }],
|
||||
ct);
|
||||
StatusCode.IsGood(write.Results[0]).ShouldBeTrue();
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() =>
|
||||
grantedEvents.HasType(grantedItem, ObjectTypeIds.AuditWriteUpdateEventType));
|
||||
|
||||
// Reported after the audit event through the same delivery, so it queues behind it on the item.
|
||||
rig.Sink.RaiseNodesAddedModelChange(EqA, AddressSpaceRealm.Uns);
|
||||
await RecordedEvents.WaitUntilAsync(() => narrowEvents.Of(narrowItem).Any(NamesEqA));
|
||||
|
||||
narrowEvents.HasType(narrowItem, ObjectTypeIds.AuditWriteUpdateEventType).ShouldBeFalse(
|
||||
"a session that may not read the node must not receive its written values through an audit event");
|
||||
|
||||
await narrowSubscription.DeleteAsync(true, ct);
|
||||
await grantedSubscription.DeleteAsync(true, ct);
|
||||
narrow.Dispose();
|
||||
granted.Dispose();
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
// ───────────────────────────── helpers ─────────────────────────────
|
||||
|
||||
private static async Task<(ISession Session, RecordedEvents Events, MonitoredItem Item, Subscription Subscription)>
|
||||
ServerObjectAsync(ServerEventRig rig, bool anonymous, CancellationToken ct, bool encrypted = false)
|
||||
{
|
||||
var session = await rig.ConnectAsync(anonymous, ct, encrypted);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var item = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
item.Status.Error.ShouldBeNull();
|
||||
return (session, events, item, subscription);
|
||||
}
|
||||
|
||||
private static bool NamesEqA(VariantCollection e) => Names(e, EqA);
|
||||
|
||||
/// <summary>Whether a model-change event's <c>Changes</c> name the node whose identifier is <paramref name="id"/>.</summary>
|
||||
private static bool Names(VariantCollection e, string id) =>
|
||||
e[RecordedEvents.ChangesField].Value is ExtensionObject[] changes
|
||||
&& changes.Any(c => c.Body is ModelChangeStructureDataType { Affected: { } affected }
|
||||
&& affected.Identifier as string == id);
|
||||
|
||||
/// <summary>Every device write fails, so the node manager reverts and audits it.</summary>
|
||||
private sealed class RejectingGateway : IOpcUaNodeWriteGateway
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public Task<NodeWriteOutcome> WriteAsync(string nodeId, object? value, AddressSpaceRealm realm, CancellationToken ct) =>
|
||||
Task.FromResult(new NodeWriteOutcome(false, "rejected by the test device"));
|
||||
}
|
||||
}
|
||||
+486
@@ -0,0 +1,486 @@
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #587 over the wire — every accepted Part 9 method call (Acknowledge, Confirm, AddComment,
|
||||
/// OneShotShelve, TimedShelve, Unshelve, Enable, Disable) on a native or scripted condition reaches a
|
||||
/// folder event item and a Server-object (<c>i=2253</c>) item as <b>exactly one</b> condition event
|
||||
/// carrying the new state.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Before #587 a client Acknowledge produced no condition event at all: the SDK's own report is gated
|
||||
/// on <c>AutoReportStateChanges</c> (never set here), and the engine's re-projection found the node
|
||||
/// already acknowledged and suppressed its own. The code comments claiming "the SDK auto-fires its own
|
||||
/// event" were wrong — and a fix that only enabled the SDK's report would still double-report a shelve
|
||||
/// (the SDK reports before any shelving state is applied, then the re-projection reports the change).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Each case runs the engine's (or, for a native alarm, the driver's) answer to the routed command as
|
||||
/// a re-projection from a pool thread, so "exactly one" covers the duplicate the re-projection could
|
||||
/// add, not only the method path. The count is read after a later <b>sentinel</b> transition arrives on
|
||||
/// the same item, so a late duplicate would already be in the queue ahead of it.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class ConditionMethodEventTests
|
||||
{
|
||||
private const NodePermissions All =
|
||||
NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe | NodePermissions.AlarmRead
|
||||
| NodePermissions.AlarmAcknowledge | NodePermissions.AlarmConfirm | NodePermissions.AlarmShelve
|
||||
| NodePermissions.MethodCall;
|
||||
|
||||
/// <summary>
|
||||
/// One accepted method call ⇒ one condition event, to the folder item and to the Server-object item,
|
||||
/// with the new state — while a session holding no grant receives none of it through the Server object.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData("Acknowledge", false, true)]
|
||||
[InlineData("Confirm", false, true)]
|
||||
[InlineData("AddComment", false, true)]
|
||||
[InlineData("OneShotShelve", false, true)]
|
||||
[InlineData("TimedShelve", false, true)]
|
||||
[InlineData("Unshelve", false, true)]
|
||||
[InlineData("Disable", false, true)]
|
||||
[InlineData("Enable", false, true)]
|
||||
// Native conditions support Acknowledge and AddComment only (the rest: A_refused_native_method_…).
|
||||
[InlineData("Acknowledge", true, true)]
|
||||
[InlineData("AddComment", true, true)]
|
||||
// Flag off: the server-wide AlarmAck role admits the call; delivery is ungated.
|
||||
[InlineData("Acknowledge", false, false)]
|
||||
[InlineData("Acknowledge", true, false)]
|
||||
[InlineData("OneShotShelve", false, false)]
|
||||
public async Task An_accepted_method_call_reports_exactly_one_condition_event(string method, bool native, bool aclOn)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync($"CondMethod{method}{native}{aclOn}", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(aclOn, [Key()], ClusterGrant(KeyPrincipal, "acl-key", All));
|
||||
rig.MaterialiseNative();
|
||||
rig.MaterialiseScripted();
|
||||
|
||||
var current = PreState(method, native);
|
||||
WireSimulatedOwners(rig, native, () => current, s => current = s);
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var folderItem = RecordedEvents.AddEventItem(subscription, UnsNodeId(session, EqA));
|
||||
var serverItem = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
folderItem.Status.Error.ShouldBeNull();
|
||||
|
||||
using var ungranted = await rig.ConnectAsync(anonymous: true, ct);
|
||||
var ungrantedEvents = new RecordedEvents();
|
||||
var ungrantedSubscription = await RecordedEvents.SubscribeAsync(ungranted, ungrantedEvents, ct);
|
||||
var ungrantedItem = RecordedEvents.AddEventItem(ungrantedSubscription, ObjectIds.Server);
|
||||
await ungrantedSubscription.ApplyChangesAsync(ct);
|
||||
|
||||
var condition = ConditionNodeId(session, native);
|
||||
|
||||
// Every transition the test drives also moves the simulated owner's state, so its re-projections
|
||||
// carry the node's current message.
|
||||
void ProjectAs(string message)
|
||||
{
|
||||
current = current with { Message = message };
|
||||
rig.Project(native, current);
|
||||
}
|
||||
|
||||
ProjectAs("base");
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(folderItem, "base") && events.Has(serverItem, "base"));
|
||||
|
||||
await CallAsync(session, condition, method, events.LatestEventId(folderItem, "base"), ct);
|
||||
await rig.DrainProjectionsAsync();
|
||||
|
||||
ProjectAs("sentinel");
|
||||
await RecordedEvents.WaitUntilAsync(() =>
|
||||
events.Has(folderItem, "sentinel") && events.Has(serverItem, "sentinel")
|
||||
&& (aclOn || ungrantedEvents.Has(ungrantedItem, "sentinel")));
|
||||
|
||||
foreach (var item in new[] { folderItem, serverItem })
|
||||
{
|
||||
var baseEvents = events.ConditionEvents(item, "base");
|
||||
baseEvents.Count.ShouldBe(2, $"the base transition plus exactly one {method} event ({item.StartNodeId})");
|
||||
AssertNewState(method, native, baseEvents[1]);
|
||||
}
|
||||
|
||||
if (aclOn)
|
||||
{
|
||||
ungrantedEvents.Of(ungrantedItem).ShouldBeEmpty(
|
||||
"a session with no grant receives no condition event through the Server object");
|
||||
}
|
||||
else
|
||||
{
|
||||
ungrantedEvents.ConditionEvents(ungrantedItem, "base").Count.ShouldBe(2, "flag off: ungated, still once");
|
||||
}
|
||||
|
||||
await ungrantedSubscription.DeleteAsync(true, ct);
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A native condition supports Acknowledge and AddComment only: Confirm, the shelve methods and
|
||||
/// Disable are refused, report nothing, and leave the condition Confirmed and Unshelved.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Native Confirm/Shelve used to be accepted and applied locally — per node, never upstream, never to the
|
||||
/// redundant peer, and undone by the driver projection (a one-shot shelve silently, a timed one never
|
||||
/// expiring). Unshelve on an unshelved condition is refused by the SDK itself before our veto runs.
|
||||
/// </remarks>
|
||||
[Theory]
|
||||
[InlineData("Confirm")]
|
||||
[InlineData("OneShotShelve")]
|
||||
[InlineData("TimedShelve")]
|
||||
[InlineData("Unshelve")]
|
||||
[InlineData("Disable")]
|
||||
public async Task A_refused_native_method_reports_no_condition_event(string method)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync($"CondMethodRefused{method}", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(true, [Key()], ClusterGrant(KeyPrincipal, "acl-key", All));
|
||||
rig.MaterialiseNative();
|
||||
var current = PreState(method, native: true);
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var folderItem = RecordedEvents.AddEventItem(subscription, UnsNodeId(session, EqA));
|
||||
var serverItem = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
rig.Project(native: true, current = current with { Message = "base" });
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(serverItem, "base"));
|
||||
|
||||
var refused = await Should.ThrowAsync<ServiceResultException>(
|
||||
CallAsync(session, ConditionNodeId(session, native: true), method, events.LatestEventId(serverItem, "base"), ct));
|
||||
if (method != "Unshelve") refused.StatusCode.ShouldBe(StatusCodes.BadNotSupported);
|
||||
|
||||
rig.Project(native: true, current with { Message = "sentinel" });
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(folderItem, "sentinel") && events.Has(serverItem, "sentinel"));
|
||||
events.ConditionEvents(folderItem, "base").Count.ShouldBe(1);
|
||||
events.ConditionEvents(serverItem, "base").Count.ShouldBe(1);
|
||||
var sentinel = events.ConditionEvents(serverItem, "sentinel")[0];
|
||||
sentinel[RecordedEvents.ConfirmedField].Value.ShouldBe(true);
|
||||
sentinel[RecordedEvents.ShelvingField].Value.ShouldBe(ObjectIds.ShelvedStateMachineType_Unshelved);
|
||||
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Materialising N conditions while a Server-object item is live costs N monitored-bookkeeping visits,
|
||||
/// not N²/2 — and every one of them still ends up marked monitored (#586 review, performance note).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Materialising_conditions_under_a_live_server_object_item_visits_each_once()
|
||||
{
|
||||
const int n = 40;
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync("CondMonitoredCost", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.EnsureEquipment();
|
||||
using var session = await rig.ConnectAsync(anonymous: true, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
var before = rig.Nm.AllEventsMonitoringConditionVisits;
|
||||
for (var i = 0; i < n; i++)
|
||||
rig.Sink.MaterialiseAlarmCondition($"bulk-{i}", EqA, $"bulk {i}", "AlarmCondition", 500, AddressSpaceRealm.Uns);
|
||||
|
||||
(rig.Nm.AllEventsMonitoringConditionVisits - before).ShouldBe(n);
|
||||
for (var i = 0; i < n; i++)
|
||||
rig.Nm.ConditionEventsMonitored($"bulk-{i}", AddressSpaceRealm.Uns).ShouldBeTrue();
|
||||
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
for (var i = 0; i < n; i++)
|
||||
rig.Nm.ConditionEventsMonitored($"bulk-{i}", AddressSpaceRealm.Uns).ShouldBeFalse(
|
||||
"the last item's delete still gives every unit back through the full reconcile");
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// An OPC UA Acknowledge on a native condition reports exactly one event with <c>ConfirmedState</c> still
|
||||
/// true, and the condition stays Confirmed through the driver's echo, a clear and a re-activation.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The SDK's Acknowledge sets Confirmed false on any condition that supports Confirm, and nothing resets
|
||||
/// it for a native alarm — ScadaBridge maps <c>ConfirmedState</c> into its alarm state, so every native
|
||||
/// alarm would read Unconfirmed after its first HMI ack (review of <c>167a48bf</c>).
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public async Task A_native_acknowledge_keeps_the_condition_confirmed_through_clear_and_reactivation()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync("CondNativeAckConfirmed", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(true, [Key()], ClusterGrant(KeyPrincipal, "acl-key", All));
|
||||
rig.MaterialiseNative();
|
||||
rig.Nm.NativeAlarmAckRouter = _ => { };
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var item = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
rig.Project(true, NativeState(active: true, acked: false, "base"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(item, "base"));
|
||||
|
||||
await CallAsync(session, ConditionNodeId(session, true), "Acknowledge", events.LatestEventId(item, "base"), ct);
|
||||
rig.Project(true, NativeState(active: true, acked: true, "base")); // the driver's echo of the ack
|
||||
rig.Project(true, NativeState(active: false, acked: true, "cleared"));
|
||||
rig.Project(true, NativeState(active: true, acked: false, "reactivated"));
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(item, "reactivated"));
|
||||
|
||||
var baseEvents = events.ConditionEvents(item, "base");
|
||||
baseEvents.Count.ShouldBe(2, "the base transition plus the one ack event; the echo is no delta");
|
||||
baseEvents[1][RecordedEvents.AckedField].Value.ShouldBe(true);
|
||||
foreach (var e in baseEvents.Concat(events.ConditionEvents(item, "cleared")).Concat(events.ConditionEvents(item, "reactivated")))
|
||||
e[RecordedEvents.ConfirmedField].Value.ShouldBe(true);
|
||||
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A scripted TimedShelve expires: the engine's expiry comes back as an Unshelved re-projection, which
|
||||
/// reports exactly one unshelve event after the one shelve event.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The method path applies the shelve with <c>shelvingTime 0</c>, as the projection does, so the expiry is
|
||||
/// the engine's (computed from the command's <c>UnshelveAtUtc</c>), never the SDK timer's — a node timer
|
||||
/// would only route an Unshelve back to the engine anyway. The stand-in engine here schedules its expiry
|
||||
/// from the real <c>UnshelveAtUtc</c> the method handler computed.
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public async Task A_scripted_timed_shelve_expires_with_exactly_one_unshelve_event()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync("CondTimedExpiry", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(true, [Key()], ClusterGrant(KeyPrincipal, "acl-key", All));
|
||||
rig.MaterialiseScripted();
|
||||
var current = PreState("TimedShelve", native: false) with { Message = "base" };
|
||||
Task? expiry = null;
|
||||
rig.Nm.AlarmCommandRouter = cmd =>
|
||||
{
|
||||
if (cmd.Operation != "TimedShelve") return;
|
||||
current = current with { Shelving = AlarmShelvingKind.Timed };
|
||||
rig.ProjectLater(native: false, current);
|
||||
var unshelveAt = cmd.UnshelveAtUtc!.Value;
|
||||
expiry = Task.Run(async () =>
|
||||
{
|
||||
var wait = unshelveAt - DateTime.UtcNow;
|
||||
if (wait > TimeSpan.Zero) await Task.Delay(wait);
|
||||
current = current with { Shelving = AlarmShelvingKind.Unshelved };
|
||||
rig.Project(native: false, current);
|
||||
});
|
||||
};
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var item = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
rig.Project(false, current);
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(item, "base"));
|
||||
|
||||
await CallAsync(session, ConditionNodeId(session, false), "TimedShelve", null, ct, shelvingTimeMs: 1000);
|
||||
await rig.DrainProjectionsAsync();
|
||||
(expiry is not null).ShouldBeTrue("the TimedShelve reached the engine");
|
||||
await expiry!;
|
||||
|
||||
rig.Project(false, current = current with { Message = "sentinel" });
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(item, "sentinel"));
|
||||
|
||||
var shelving = events.ConditionEvents(item, "base").Select(e => e[RecordedEvents.ShelvingField].Value).ToList();
|
||||
shelving.ShouldBe(
|
||||
[ObjectIds.ShelvedStateMachineType_Unshelved, ObjectIds.ShelvedStateMachineType_TimedShelved,
|
||||
ObjectIds.ShelvedStateMachineType_Unshelved],
|
||||
"base, then exactly one shelve event, then exactly one expiry event");
|
||||
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Acknowledging an acknowledged condition reports nothing, and does not flip a scripted condition's
|
||||
/// Confirmed out of step with its engine.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The SDK accepts a re-ack and sets Confirmed false again; the engine ignores it. The scripted leg
|
||||
/// acknowledges and confirms first, so the flip would be visible on the sentinel.
|
||||
/// </remarks>
|
||||
[Theory]
|
||||
[InlineData(false)]
|
||||
[InlineData(true)]
|
||||
public async Task A_re_acknowledge_reports_nothing_and_changes_nothing(bool native)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync($"CondReAck{native}", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(true, [Key()], ClusterGrant(KeyPrincipal, "acl-key", All));
|
||||
rig.MaterialiseNative();
|
||||
rig.MaterialiseScripted();
|
||||
var current = PreState("Acknowledge", native) with { Message = "base" };
|
||||
WireSimulatedOwners(rig, native, () => current, s => current = s);
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var item = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
var condition = ConditionNodeId(session, native);
|
||||
rig.Project(native, current);
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(item, "base"));
|
||||
|
||||
await CallAsync(session, condition, "Acknowledge", events.LatestEventId(item, "base"), ct);
|
||||
await rig.DrainProjectionsAsync();
|
||||
var expected = 2;
|
||||
if (!native)
|
||||
{
|
||||
await RecordedEvents.WaitUntilAsync(() => events.ConditionEvents(item, "base").Count == 2);
|
||||
await CallAsync(session, condition, "Confirm", events.LatestEventId(item, "base"), ct);
|
||||
await rig.DrainProjectionsAsync();
|
||||
expected = 3;
|
||||
}
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() => events.ConditionEvents(item, "base").Count == expected);
|
||||
await CallAsync(session, condition, "Acknowledge", events.LatestEventId(item, "base"), ct);
|
||||
await rig.DrainProjectionsAsync();
|
||||
|
||||
rig.Project(native, current = current with { Message = "sentinel" });
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(item, "sentinel"));
|
||||
|
||||
events.ConditionEvents(item, "base").Count.ShouldBe(expected, "the re-ack must report nothing");
|
||||
var sentinel = events.ConditionEvents(item, "sentinel");
|
||||
sentinel.Count.ShouldBe(1);
|
||||
sentinel[0][RecordedEvents.AckedField].Value.ShouldBe(true);
|
||||
sentinel[0][RecordedEvents.ConfirmedField].Value.ShouldBe(true, "the re-ack must not unconfirm the node");
|
||||
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
// ───────────────────────────── helpers ─────────────────────────────
|
||||
|
||||
private static AlarmConditionSnapshot NativeState(bool active, bool acked, string message) => new(
|
||||
Active: active, Acknowledged: acked, Confirmed: true, Enabled: true,
|
||||
Shelving: AlarmShelvingKind.Unshelved, Severity: 700, Message: message);
|
||||
|
||||
/// <summary>The condition state a case starts from — one in which its method is accepted.</summary>
|
||||
private static AlarmConditionSnapshot PreState(string method, bool native)
|
||||
{
|
||||
var s = new AlarmConditionSnapshot(
|
||||
Active: true, Acknowledged: false, Confirmed: native, Enabled: true,
|
||||
Shelving: AlarmShelvingKind.Unshelved, Severity: 700, Message: "pre");
|
||||
if (native) return s; // the driver projection's constants: Confirmed, Unshelved
|
||||
|
||||
return method switch
|
||||
{
|
||||
"Confirm" => s with { Acknowledged = true, Confirmed = false },
|
||||
"Unshelve" => s with { Shelving = AlarmShelvingKind.OneShot },
|
||||
"Enable" => s with { Enabled = false },
|
||||
_ => s,
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Stands in for the scripted engine and the native driver: each routed command comes back as a
|
||||
/// re-projection of the state it produces, from a pool thread.
|
||||
/// </summary>
|
||||
private static void WireSimulatedOwners(
|
||||
ServerEventRig rig, bool native, Func<AlarmConditionSnapshot> current, Action<AlarmConditionSnapshot> set)
|
||||
{
|
||||
rig.Nm.NativeAlarmAckRouter = _ =>
|
||||
{
|
||||
// NativeAlarmProjector's answer to the driver's ack echo: Acked, with its Confirmed/Unshelved constants.
|
||||
var next = current() with { Acknowledged = true, Confirmed = true, Shelving = AlarmShelvingKind.Unshelved };
|
||||
set(next);
|
||||
rig.ProjectLater(native: true, next);
|
||||
};
|
||||
rig.Nm.AlarmCommandRouter = cmd =>
|
||||
{
|
||||
if (native || cmd.AlarmId != ScriptedAlarmId) return; // the engine owns only scripted alarms
|
||||
var s = current();
|
||||
var next = cmd.Operation switch
|
||||
{
|
||||
"Acknowledge" => s with { Acknowledged = true },
|
||||
"Confirm" => s with { Confirmed = true },
|
||||
"OneShotShelve" => s with { Shelving = AlarmShelvingKind.OneShot },
|
||||
"TimedShelve" => s with { Shelving = AlarmShelvingKind.Timed },
|
||||
"Unshelve" => s with { Shelving = AlarmShelvingKind.Unshelved },
|
||||
"Enable" => s with { Enabled = true },
|
||||
"Disable" => s with { Enabled = false },
|
||||
_ => s, // AddComment: no field of the snapshot changes
|
||||
};
|
||||
set(next);
|
||||
rig.ProjectLater(native: false, next);
|
||||
};
|
||||
}
|
||||
|
||||
private static void AssertNewState(string method, bool native, VariantCollection e)
|
||||
{
|
||||
// A native condition is never Unconfirmed, whatever the method (the SDK's ack would make it so).
|
||||
if (native) e[RecordedEvents.ConfirmedField].Value.ShouldBe(true);
|
||||
switch (method)
|
||||
{
|
||||
case "Acknowledge": e[RecordedEvents.AckedField].Value.ShouldBe(true); break;
|
||||
case "Confirm": e[RecordedEvents.ConfirmedField].Value.ShouldBe(true); break;
|
||||
case "AddComment": (e[RecordedEvents.CommentField].Value as LocalizedText)?.Text.ShouldBe("note"); break;
|
||||
case "OneShotShelve": e[RecordedEvents.ShelvingField].Value.ShouldBe(ObjectIds.ShelvedStateMachineType_OneShotShelved); break;
|
||||
case "TimedShelve": e[RecordedEvents.ShelvingField].Value.ShouldBe(ObjectIds.ShelvedStateMachineType_TimedShelved); break;
|
||||
case "Unshelve": e[RecordedEvents.ShelvingField].Value.ShouldBe(ObjectIds.ShelvedStateMachineType_Unshelved); break;
|
||||
case "Disable": e[RecordedEvents.EnabledField].Value.ShouldBe(false); break;
|
||||
case "Enable": e[RecordedEvents.EnabledField].Value.ShouldBe(true); break;
|
||||
default: throw new ArgumentOutOfRangeException(nameof(method), method, null);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Calls one Part 9 method on the condition by its type's method id (the SDK resolves it against the
|
||||
/// condition's method declarations — the shelve methods included); throws on a bad result.
|
||||
/// </summary>
|
||||
private static async Task CallAsync(
|
||||
ISession session, NodeId condition, string method, byte[]? eventId, CancellationToken ct, double shelvingTimeMs = 60_000)
|
||||
{
|
||||
var (methodId, args) = method switch
|
||||
{
|
||||
"Acknowledge" => (MethodIds.AcknowledgeableConditionType_Acknowledge, new object[] { eventId!, new LocalizedText("ack") }),
|
||||
"Confirm" => (MethodIds.AcknowledgeableConditionType_Confirm, [eventId!, new LocalizedText("confirm")]),
|
||||
"AddComment" => (MethodIds.ConditionType_AddComment, [eventId!, new LocalizedText("note")]),
|
||||
"Enable" => (MethodIds.ConditionType_Enable, []),
|
||||
"Disable" => (MethodIds.ConditionType_Disable, []),
|
||||
"OneShotShelve" => (MethodIds.AlarmConditionType_ShelvingState_OneShotShelve, []),
|
||||
"TimedShelve" => (MethodIds.AlarmConditionType_ShelvingState_TimedShelve, [shelvingTimeMs]),
|
||||
"Unshelve" => (MethodIds.AlarmConditionType_ShelvingState_Unshelve, []),
|
||||
_ => throw new ArgumentOutOfRangeException(nameof(method), method, null),
|
||||
};
|
||||
await session.CallAsync(condition, methodId, ct, args);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,373 @@
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Reflection;
|
||||
using System.Text;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Opc.Ua.Configuration;
|
||||
using Shouldly;
|
||||
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.Host.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// A booted server for the Server-object event tests (Gitea #587 / #588): one native condition on a raw
|
||||
/// device wired to two equipment notifiers, one scripted condition under <see cref="EqA"/>, an API-key
|
||||
/// principal and anonymous, optionally an encrypted endpoint and server auditing.
|
||||
/// </summary>
|
||||
internal sealed class ServerEventRig : IAsyncDisposable
|
||||
{
|
||||
public const string Cluster = "SITE-A";
|
||||
public const string RawDeviceFolder = "Plant/Modbus/dev1";
|
||||
public const string RawAlarmPath = "Plant/Modbus/dev1/temp_hi";
|
||||
public const string ScriptedAlarmId = "scripted-eqa-hi";
|
||||
public const string EqA = "EqA";
|
||||
public const string EqB = "EqB";
|
||||
public const string KeyPrincipal = "event-probe";
|
||||
public const string KeyId = "k-server-events";
|
||||
|
||||
private readonly string _pkiRoot;
|
||||
private readonly List<Task> _pending = [];
|
||||
private OpcUaApplicationHost _host = null!;
|
||||
private OtOpcUaSdkServer _server = null!;
|
||||
private int _port;
|
||||
|
||||
private ServerEventRig(string pkiRoot) => _pkiRoot = pkiRoot;
|
||||
|
||||
public OtOpcUaNodeManager Nm => _server.NodeManager!;
|
||||
|
||||
public SdkAddressSpaceSink Sink { get; private set; } = null!;
|
||||
|
||||
public static async Task<ServerEventRig> StartAsync(string name, bool encrypted, CancellationToken ct)
|
||||
{
|
||||
var rig = new ServerEventRig(Path.Combine(Path.GetTempPath(), $"otopcua-{name}-{Guid.NewGuid():N}"));
|
||||
rig._port = AllocateFreePort();
|
||||
rig._server = new OtOpcUaSdkServer();
|
||||
rig._host = new OpcUaApplicationHost(
|
||||
new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = $"OtOpcUa.{name}",
|
||||
ApplicationUri = $"urn:OtOpcUa.{name}.{Guid.NewGuid():N}",
|
||||
OpcUaPort = rig._port,
|
||||
PublicHostname = "127.0.0.1",
|
||||
PkiStoreRoot = rig._pkiRoot,
|
||||
EnabledSecurityProfiles = encrypted
|
||||
? [OpcUaSecurityProfile.None, OpcUaSecurityProfile.Basic256Sha256SignAndEncrypt]
|
||||
: [OpcUaSecurityProfile.None],
|
||||
AutoAcceptUntrustedClientCertificates = true,
|
||||
},
|
||||
NullLogger<OpcUaApplicationHost>.Instance,
|
||||
new KeyPresentingAuthenticator());
|
||||
|
||||
await rig._host.StartAsync(rig._server, ct);
|
||||
rig.Sink = new SdkAddressSpaceSink(rig.Nm);
|
||||
return rig;
|
||||
}
|
||||
|
||||
/// <summary>Wires the authorizer over the given keys and grants, enforcing or not.</summary>
|
||||
public DataPlaneAuthorityState Enforce(bool enabled, ApiKey[] keys, params NodeAcl[] grants)
|
||||
{
|
||||
var state = new DataPlaneAuthorityState();
|
||||
state.Apply(Cluster, "rev-1", keys, grants, ScopePaths());
|
||||
Nm.DataPlaneAuthority = state;
|
||||
Nm.Authorizer = new DataPlaneAuthorizer(
|
||||
state, new TriePermissionEvaluator(state.Tries), enabled, NullLogger.Instance, TimeProvider.System);
|
||||
return state;
|
||||
}
|
||||
|
||||
public static ApiKey Key() => new()
|
||||
{
|
||||
ApiKeyId = KeyId,
|
||||
Name = KeyId,
|
||||
SecretHash = "hash",
|
||||
Principal = KeyPrincipal,
|
||||
Enabled = true,
|
||||
CreatedUtc = DateTime.UtcNow.AddDays(-1),
|
||||
};
|
||||
|
||||
public static NodeAcl ClusterGrant(string principal, string id, NodePermissions permissions) => new()
|
||||
{
|
||||
NodeAclId = id,
|
||||
ClusterId = Cluster,
|
||||
Principal = principal,
|
||||
ScopeKind = NodeAclScopeKind.Cluster,
|
||||
PermissionFlags = permissions,
|
||||
};
|
||||
|
||||
public static NodeAcl EquipmentGrant(string equipment, NodePermissions permissions) => new()
|
||||
{
|
||||
NodeAclId = $"acl-{equipment}",
|
||||
ClusterId = Cluster,
|
||||
Principal = OpcUaPrincipals.Anonymous,
|
||||
ScopeKind = NodeAclScopeKind.Equipment,
|
||||
ScopeId = $"eq-{equipment}",
|
||||
PermissionFlags = permissions,
|
||||
};
|
||||
|
||||
private static IReadOnlyDictionary<AclScopeRef, IReadOnlyList<NodeAclPath>> ScopePaths() =>
|
||||
new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>
|
||||
{
|
||||
[new AclScopeRef(NodeAclScopeKind.Equipment, $"eq-{EqA}")] = [new NodeAclPath([NodeAclRealms.Uns, EqA])],
|
||||
[new AclScopeRef(NodeAclScopeKind.Equipment, $"eq-{EqB}")] = [new NodeAclPath([NodeAclRealms.Uns, EqB])],
|
||||
};
|
||||
|
||||
/// <summary>The native condition, wired as an event source of both equipment folders.</summary>
|
||||
public void MaterialiseNative()
|
||||
{
|
||||
Sink.EnsureFolder(RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
|
||||
Sink.MaterialiseAlarmCondition(
|
||||
RawAlarmPath, RawDeviceFolder, "temp_hi", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true);
|
||||
EnsureEquipment();
|
||||
Sink.WireAlarmNotifiers(RawAlarmPath, AddressSpaceRealm.Raw, [EqA, EqB], AddressSpaceRealm.Uns);
|
||||
}
|
||||
|
||||
/// <summary>The scripted condition under <see cref="EqA"/>.</summary>
|
||||
public void MaterialiseScripted()
|
||||
{
|
||||
EnsureEquipment();
|
||||
Sink.MaterialiseAlarmCondition(
|
||||
ScriptedAlarmId, EqA, "hi", "AlarmCondition", 500, AddressSpaceRealm.Uns, isNative: false);
|
||||
}
|
||||
|
||||
public void EnsureEquipment()
|
||||
{
|
||||
Sink.EnsureFolder(EqA, parentNodeId: null, displayName: "A", realm: AddressSpaceRealm.Uns);
|
||||
Sink.EnsureFolder(EqB, parentNodeId: null, displayName: "B", realm: AddressSpaceRealm.Uns);
|
||||
}
|
||||
|
||||
/// <summary>Projects <paramref name="state"/> onto a condition, as the engine or a driver would.</summary>
|
||||
public void Project(bool native, AlarmConditionSnapshot state) =>
|
||||
Sink.WriteAlarmCondition(
|
||||
native ? RawAlarmPath : ScriptedAlarmId,
|
||||
state,
|
||||
DateTime.UtcNow,
|
||||
native ? AddressSpaceRealm.Raw : AddressSpaceRealm.Uns);
|
||||
|
||||
/// <summary>
|
||||
/// Re-projects <paramref name="state"/> from a pool thread, the way the engine's or the driver's answer
|
||||
/// to a routed method call arrives — it contends for the node manager's lock like the real one does.
|
||||
/// </summary>
|
||||
public void ProjectLater(bool native, AlarmConditionSnapshot state)
|
||||
{
|
||||
lock (_pending) _pending.Add(Task.Run(() => Project(native, state)));
|
||||
}
|
||||
|
||||
/// <summary>Waits for every <see cref="ProjectLater"/> re-projection to have run.</summary>
|
||||
public async Task DrainProjectionsAsync()
|
||||
{
|
||||
Task[] pending;
|
||||
lock (_pending) pending = [.. _pending];
|
||||
await Task.WhenAll(pending);
|
||||
}
|
||||
|
||||
/// <summary>The condition's NodeId in the session's namespaces.</summary>
|
||||
public static NodeId ConditionNodeId(ISession session, bool native) => native
|
||||
? new NodeId(RawAlarmPath, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri))
|
||||
: new NodeId(ScriptedAlarmId, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.UnsNamespaceUri));
|
||||
|
||||
public static NodeId UnsNodeId(ISession session, string id)
|
||||
{
|
||||
var ns = (ushort)session.NamespaceUris.GetIndex(V3NodeIds.UnsNamespaceUri);
|
||||
ns.ShouldBeGreaterThan((ushort)0);
|
||||
return new NodeId(id, ns);
|
||||
}
|
||||
|
||||
/// <summary>Switches the server's <c>Auditing</c> flag (no configuration seam reaches it in this repo).</summary>
|
||||
public void SetAuditing(bool on)
|
||||
{
|
||||
var server = typeof(Opc.Ua.Server.CustomNodeManager2)
|
||||
.GetProperty("Server", BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public)!
|
||||
.GetValue(Nm)!;
|
||||
server.GetType()
|
||||
.GetProperty("Auditing", BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic)!
|
||||
.GetSetMethod(nonPublic: true)!
|
||||
.Invoke(server, [on]);
|
||||
}
|
||||
|
||||
public async Task<ISession> ConnectAsync(bool anonymous, CancellationToken ct, bool encrypted = false)
|
||||
{
|
||||
var pki = TestClientSecurity.AllocatePkiRoot();
|
||||
var security = TestClientSecurity.Build(pki);
|
||||
security.ApplicationCertificate.SubjectName = "CN=OtOpcUa.ServerEventClient";
|
||||
var appConfig = new ApplicationConfiguration
|
||||
{
|
||||
ApplicationName = "OtOpcUa.ServerEventClient",
|
||||
ApplicationUri = $"urn:OtOpcUa.ServerEventClient.{Guid.NewGuid():N}",
|
||||
ApplicationType = ApplicationType.Client,
|
||||
SecurityConfiguration = security,
|
||||
ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 60_000 },
|
||||
};
|
||||
await appConfig.ValidateAsync(ApplicationType.Client, ct);
|
||||
appConfig.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true;
|
||||
|
||||
if (encrypted)
|
||||
{
|
||||
#pragma warning disable CS0618 // the telemetry-context overload is not needed for a test client
|
||||
var app = new ApplicationInstance { ApplicationName = appConfig.ApplicationName, ApplicationType = ApplicationType.Client };
|
||||
#pragma warning restore CS0618
|
||||
app.ApplicationConfiguration = appConfig;
|
||||
await app.CheckApplicationInstanceCertificatesAsync(silent: true, lifeTimeInMonths: null, ct);
|
||||
}
|
||||
|
||||
var endpoint = await CoreClientUtils.SelectEndpointAsync(
|
||||
appConfig, $"opc.tcp://127.0.0.1:{_port}/OtOpcUa", encrypted, DefaultTelemetry.Create(_ => { }), ct);
|
||||
if (encrypted) endpoint!.SecurityMode.ShouldBe(MessageSecurityMode.SignAndEncrypt);
|
||||
|
||||
var identity = anonymous
|
||||
? new UserIdentity(new AnonymousIdentityToken())
|
||||
: new UserIdentity(new UserNameIdentityToken
|
||||
{
|
||||
UserName = "apikey",
|
||||
DecryptedPassword = Encoding.UTF8.GetBytes($"otua_{KeyId}_secret"),
|
||||
});
|
||||
|
||||
return await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(
|
||||
appConfig,
|
||||
new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(appConfig)),
|
||||
updateBeforeConnect: false,
|
||||
checkDomain: false,
|
||||
sessionName: "ServerEventRig",
|
||||
sessionTimeout: 60_000,
|
||||
identity: identity,
|
||||
preferredLocales: null,
|
||||
ct: ct);
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _host.DisposeAsync();
|
||||
if (Directory.Exists(_pkiRoot))
|
||||
{
|
||||
try { Directory.Delete(_pkiRoot, recursive: true); }
|
||||
catch (IOException) { /* best effort */ }
|
||||
}
|
||||
}
|
||||
|
||||
private static int AllocateFreePort()
|
||||
{
|
||||
var listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
listener.Stop();
|
||||
return port;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Admits <c>apikey</c> as <see cref="KeyPrincipal"/> carrying the two server-wide roles (used only
|
||||
/// while enforcement is off); anonymous falls through to the <c>anonymous</c> principal.
|
||||
/// </summary>
|
||||
private sealed class KeyPresentingAuthenticator : IOpcUaUserAuthenticator
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public Task<OpcUaUserAuthResult> AuthenticateUserNameAsync(string username, string password, CancellationToken ct) =>
|
||||
Task.FromResult(
|
||||
string.Equals(username, "apikey", StringComparison.Ordinal)
|
||||
&& password.StartsWith($"otua_{KeyId}_", StringComparison.Ordinal)
|
||||
? OpcUaUserAuthResult.AllowKey(
|
||||
KeyPrincipal, [OpcUaDataPlaneRoles.AlarmAck, OpcUaDataPlaneRoles.WriteOperate], KeyId)
|
||||
: OpcUaUserAuthResult.Deny("Invalid API key."));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Records every event each monitored item received, in order, with the fields the tests select.</summary>
|
||||
internal sealed class RecordedEvents
|
||||
{
|
||||
// The select clauses, in order — see AddEventItem.
|
||||
public const int EventIdField = 0;
|
||||
public const int MessageField = 1;
|
||||
public const int EventTypeField = 2;
|
||||
public const int AckedField = 3;
|
||||
public const int ConfirmedField = 4;
|
||||
public const int EnabledField = 5;
|
||||
public const int ShelvingField = 6;
|
||||
public const int CommentField = 7;
|
||||
public const int ChangesField = 8;
|
||||
public const int SourceNodeField = 9;
|
||||
|
||||
private readonly object _gate = new();
|
||||
private readonly List<(uint Handle, VariantCollection Fields)> _events = [];
|
||||
|
||||
public void OnEvents(Subscription subscription, EventNotificationList notification, IList<string> stringTable)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
foreach (var e in notification.Events) _events.Add((e.ClientHandle, e.EventFields));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Every event <paramref name="item"/> received, in arrival order.</summary>
|
||||
public IReadOnlyList<VariantCollection> Of(MonitoredItem item)
|
||||
{
|
||||
lock (_gate) return _events.Where(e => e.Handle == item.ClientHandle).Select(e => e.Fields).ToList();
|
||||
}
|
||||
|
||||
/// <summary>The condition events (they carry <c>AckedState/Id</c>; audit and refresh events do not).</summary>
|
||||
public IReadOnlyList<VariantCollection> ConditionEvents(MonitoredItem item, string message) =>
|
||||
Of(item).Where(f => f[AckedField].Value is bool && Message(f) == message).ToList();
|
||||
|
||||
public bool Has(MonitoredItem item, string message) => Of(item).Any(f => Message(f) == message);
|
||||
|
||||
public bool HasType(MonitoredItem item, NodeId type) => Of(item).Any(f => f[EventTypeField].Value as NodeId == type);
|
||||
|
||||
public byte[] LatestEventId(MonitoredItem item, string message) =>
|
||||
(byte[])Of(item).Last(f => Message(f) == message && f[AckedField].Value is bool)[EventIdField].Value;
|
||||
|
||||
public static string? Message(VariantCollection f) => (f[MessageField].Value as LocalizedText)?.Text;
|
||||
|
||||
public static async Task<Subscription> SubscribeAsync(ISession session, RecordedEvents events, CancellationToken ct)
|
||||
{
|
||||
var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
|
||||
subscription.FastEventCallback = events.OnEvents;
|
||||
session.AddSubscription(subscription);
|
||||
await subscription.CreateAsync(ct);
|
||||
return subscription;
|
||||
}
|
||||
|
||||
public static MonitoredItem AddEventItem(Subscription subscription, NodeId source)
|
||||
{
|
||||
var filter = new EventFilter();
|
||||
filter.AddSelectClause(ObjectTypes.BaseEventType, BrowseNames.EventId);
|
||||
filter.AddSelectClause(ObjectTypes.BaseEventType, BrowseNames.Message);
|
||||
filter.AddSelectClause(ObjectTypes.BaseEventType, BrowseNames.EventType);
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand(
|
||||
ObjectTypeIds.AcknowledgeableConditionType, [BrowseNames.AckedState, BrowseNames.Id]));
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand(
|
||||
ObjectTypeIds.AcknowledgeableConditionType, [BrowseNames.ConfirmedState, BrowseNames.Id]));
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand(
|
||||
ObjectTypeIds.ConditionType, [BrowseNames.EnabledState, BrowseNames.Id]));
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand(
|
||||
ObjectTypeIds.AlarmConditionType, [BrowseNames.ShelvingState, BrowseNames.CurrentState, BrowseNames.Id]));
|
||||
filter.AddSelectClause(ObjectTypeIds.ConditionType, BrowseNames.Comment);
|
||||
filter.AddSelectClause(ObjectTypeIds.GeneralModelChangeEventType, BrowseNames.Changes);
|
||||
filter.AddSelectClause(ObjectTypes.BaseEventType, BrowseNames.SourceNode);
|
||||
|
||||
var item = new MonitoredItem(subscription.DefaultItem)
|
||||
{
|
||||
StartNodeId = source,
|
||||
AttributeId = Attributes.EventNotifier,
|
||||
Filter = filter,
|
||||
QueueSize = 1000,
|
||||
SamplingInterval = 0,
|
||||
};
|
||||
subscription.AddItem(item);
|
||||
return item;
|
||||
}
|
||||
|
||||
public static async Task WaitUntilAsync(Func<bool> condition)
|
||||
{
|
||||
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
if (condition()) return;
|
||||
await Task.Delay(50);
|
||||
}
|
||||
condition().ShouldBeTrue("condition not met within timeout");
|
||||
}
|
||||
}
|
||||
+125
@@ -0,0 +1,125 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
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.Authorization;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #588 — <see cref="OtOpcUaNodeManager.FilterModelChangeFor"/> narrows a model-change event to the
|
||||
/// changes a session may browse. Every builder in the manager emits one change, so the over-the-wire tests
|
||||
/// only ever see "whole or nothing"; this pins the multi-change copy at its seam.
|
||||
/// </summary>
|
||||
public sealed class NodeManagerModelChangeFilterTests : IDisposable
|
||||
{
|
||||
private const string Cluster = "SITE-A";
|
||||
private const string Principal = "scadabridge";
|
||||
private const string RawFolder = "Plant/Modbus/dev1";
|
||||
|
||||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||||
|
||||
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-mcfilter-{Guid.NewGuid():N}");
|
||||
|
||||
/// <summary>
|
||||
/// A session granted one equipment folder receives a copy carrying only that folder's change — same
|
||||
/// EventId, and a Message that no longer names the hidden node — while a fully granted session receives
|
||||
/// the original and an ungranted one nothing.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_multi_change_event_is_narrowed_to_the_changes_the_session_may_browse()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
try
|
||||
{
|
||||
var nm = server.NodeManager!;
|
||||
var sink = new SdkAddressSpaceSink(nm);
|
||||
sink.EnsureFolder("EqA", null, "A", AddressSpaceRealm.Uns);
|
||||
sink.EnsureFolder(RawFolder, null, "dev1", AddressSpaceRealm.Raw);
|
||||
|
||||
var e = nm.BuildNodesAddedModelChange("EqA", AddressSpaceRealm.Uns);
|
||||
var eqA = e.Changes.Value[0];
|
||||
var raw = nm.BuildNodesAddedModelChange(RawFolder, AddressSpaceRealm.Raw).Changes.Value[0];
|
||||
e.Changes.Value = [eqA, raw];
|
||||
e.Message.Value = new LocalizedText($"Nodes added under {eqA.Affected} and {raw.Affected}");
|
||||
|
||||
nm.Authorizer = Authorizer(
|
||||
new NodeAcl
|
||||
{
|
||||
NodeAclId = "acl-eqa", ClusterId = Cluster, Principal = Principal,
|
||||
ScopeKind = NodeAclScopeKind.Equipment, ScopeId = "eq-EqA",
|
||||
PermissionFlags = NodePermissions.Browse,
|
||||
},
|
||||
new NodeAcl
|
||||
{
|
||||
NodeAclId = "acl-all", ClusterId = Cluster, Principal = "everything",
|
||||
ScopeKind = NodeAclScopeKind.Cluster, PermissionFlags = NodePermissions.Browse,
|
||||
});
|
||||
|
||||
var narrowed = nm.FilterModelChangeFor(Identity(Principal), e).ShouldNotBeNull();
|
||||
narrowed.ShouldNotBeSameAs(e);
|
||||
narrowed.Changes.Value.Select(c => c.Affected).ShouldBe([eqA.Affected]);
|
||||
narrowed.EventId.Value.ShouldBe(e.EventId.Value);
|
||||
narrowed.Message.Value.Text.ShouldNotContain(RawFolder);
|
||||
|
||||
nm.FilterModelChangeFor(Identity("everything"), e).ShouldBeSameAs(e);
|
||||
nm.FilterModelChangeFor(Identity("nobody"), e).ShouldBeNull();
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
private static DataPlaneAuthorizer Authorizer(params NodeAcl[] grants)
|
||||
{
|
||||
var state = new DataPlaneAuthorityState();
|
||||
state.Apply(Cluster, "rev-1", [], grants,
|
||||
new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>
|
||||
{
|
||||
[new AclScopeRef(NodeAclScopeKind.Equipment, "eq-EqA")] = [new NodeAclPath([NodeAclRealms.Uns, "EqA"])],
|
||||
});
|
||||
return new DataPlaneAuthorizer(state, new TriePermissionEvaluator(state.Tries), true, NullLogger.Instance);
|
||||
}
|
||||
|
||||
private static RoleCarryingUserIdentity Identity(string principal) =>
|
||||
new(new UserNameIdentityToken { UserName = "apikey" }, [], principal, apiKeyId: null);
|
||||
|
||||
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
|
||||
{
|
||||
var host = new OpcUaApplicationHost(
|
||||
new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = "OtOpcUa.ModelChangeFilterTest",
|
||||
ApplicationUri = $"urn:OtOpcUa.ModelChangeFilterTest:{Guid.NewGuid():N}",
|
||||
OpcUaPort = AllocateFreePort(),
|
||||
PublicHostname = "localhost",
|
||||
PkiStoreRoot = _pkiRoot,
|
||||
},
|
||||
NullLogger<OpcUaApplicationHost>.Instance);
|
||||
|
||||
var server = new OtOpcUaSdkServer();
|
||||
await host.StartAsync(server, Ct);
|
||||
return (host, server);
|
||||
}
|
||||
|
||||
private static int AllocateFreePort()
|
||||
{
|
||||
using var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port;
|
||||
listener.Stop();
|
||||
return port;
|
||||
}
|
||||
|
||||
/// <summary>Cleans up the PKI root directory.</summary>
|
||||
public void Dispose()
|
||||
{
|
||||
if (Directory.Exists(_pkiRoot))
|
||||
{
|
||||
try { Directory.Delete(_pkiRoot, recursive: true); }
|
||||
catch (IOException) { /* best effort */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user