lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AclEventSubscriptionEnforcementTests.cs

using System.Net;
using System.Net.Sockets;
using System.Text;
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Client;
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.Host.OpcUa;
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;

/// <summary>
///     Gitea #562 over the wire — an <b>alarm</b> (event) subscription is re-authorized by both revalidation
///     triggers, exactly as a data-change subscription is: a deployment that removes its grant, and the host's
///     clock-driven <see cref="AclRevalidationTimer"/> once the session's key expires.
/// </summary>
/// <remarks>
///     <para>
///         Before #562 an event monitored item was gated at creation and never again, so an alarm feed
///         survived the revocation of the authority it was admitted under until the client disconnected.
///     </para>
///     <para>
///         Over the wire for the reason <c>AclSubscriptionEnforcementTests</c> gives: whether an event item
///         can recover the identity of the session that created it, and whether detaching it from the SDK's
///         event fan-out really stops delivery (and re-attaching really resumes it), are assumptions about SDK
///         behaviour that fail silently if wrong.
///     </para>
///     <para>
///         Every absence assertion is paired with a <b>positive control on the same event</b>: one condition is
///         wired to two equipment notifiers (or watched by two sessions), so a single transition must reach the
///         still-authorized item before the denied one's silence counts for anything.
///     </para>
/// </remarks>
public sealed class AclEventSubscriptionEnforcementTests
{
    private const string Cluster = "SITE-A";
    private const string RawDeviceFolder = "Plant/Modbus/dev1";
    private const string RawAlarmPath = "Plant/Modbus/dev1/temp_hi";
    private const string EqA = "EqA";
    private const string EqB = "EqB";
    private const string KeyPrincipal = "expiry-probe";
    private const string KeyId = "k-event-expiry";

    // [0]=EventId, [1]=Message — the message is each transition's discriminator.
    private const int MessageFieldIndex = 1;

    private static readonly DateTime Start = new(2026, 10, 7, 12, 0, 0, DateTimeKind.Utc);

    /// <summary>
    ///     The deployment axis, end to end: revoke → the alarm feed stops (while the still-granted control
    ///     item receives the same event) → restore → the feed resumes with no client action.
    /// </summary>
    /// <remarks>
    ///     The restore leg is the reason the stand-down is a detach rather than
    ///     <c>MonitoringMode.Disabled</c>: a disabled item can only be re-enabled by the client, and a machine
    ///     consumer never will.
    /// </remarks>
    [Fact]
    public async Task Revoking_then_restoring_a_grant_stops_then_resumes_a_live_alarm_subscription()
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await Rig.StartAsync("AclEventRevoke", ct);
        try
        {
            var state = rig.Enforce(EquipmentGrant(EqA), EquipmentGrant(EqB));
            rig.Materialise();

            using var session = await rig.ConnectAsync(anonymous: true, ct);
            var collector = new EventCollector();
            var subscription = await CreateSubscriptionAsync(session, collector, ct);
            var onA = AddEventItem(subscription, rig.UnsNodeId(session, EqA));
            var onB = AddEventItem(subscription, rig.UnsNodeId(session, EqB));
            await subscription.ApplyChangesAsync(ct);
            onA.Status.Error.ShouldBeNull();
            onB.Status.Error.ShouldBeNull();

            // Baseline: both items receive the transition.
            rig.Transition("t1", active: true);
            await WaitUntilAsync(() => collector.Has(onA, "t1") && collector.Has(onB, "t1"));

            // The next deployment carries only EqB's grant.
            state.Apply(Cluster, "rev-2", [], [EquipmentGrant(EqB)], ScopePaths());
            rig.AttachedEventItems.ShouldBe(1);

            rig.Transition("t2", active: false);
            // Positive control: the same event reached the still-granted item, so it was fired and published…
            await WaitUntilAsync(() => collector.Has(onB, "t2"));
            // …and settle past a publishing interval so a late copy for EqA would have arrived too.
            await Task.Delay(TimeSpan.FromMilliseconds(700), ct);
            collector.Has(onA, "t2").ShouldBeFalse("the revoked item must not receive events after the deployment");

            // A deployment that restores the grant brings the feed back on its own.
            state.Apply(Cluster, "rev-3", [], [EquipmentGrant(EqA), EquipmentGrant(EqB)], ScopePaths());
            rig.AttachedEventItems.ShouldBe(2);

            rig.Transition("t3", active: true);
            await WaitUntilAsync(() => collector.Has(onA, "t3") && collector.Has(onB, "t3"));
            collector.Has(onA, "t2").ShouldBeFalse("nothing withheld while stood down is replayed on restore");

            await subscription.DeleteAsync(true, ct);
        }
        finally { await rig.DisposeAsync(); }
    }

    /// <summary>
    ///     A client may delete an item while it is stood down: the delete succeeds, and a later deployment
    ///     that would have restored it does not resurrect it onto the notifier's fan-out.
    /// </summary>
    /// <remarks>
    ///     The SDK's own unsubscribe would try to remove an item that is no longer attached, so the node
    ///     manager short-circuits it. Asserted over the wire because the failure — an exception inside the
    ///     SDK's delete path, or a ghost item still receiving events — is only observable there.
    /// </remarks>
    [Fact]
    public async Task A_stood_down_alarm_item_can_be_deleted_and_is_not_resurrected()
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await Rig.StartAsync("AclEventDelete", ct);
        try
        {
            var state = rig.Enforce(EquipmentGrant(EqA), EquipmentGrant(EqB));
            rig.Materialise();

            using var session = await rig.ConnectAsync(anonymous: true, ct);
            var collector = new EventCollector();
            var subscription = await CreateSubscriptionAsync(session, collector, ct);
            var onA = AddEventItem(subscription, rig.UnsNodeId(session, EqA));
            var onB = AddEventItem(subscription, rig.UnsNodeId(session, EqB));
            await subscription.ApplyChangesAsync(ct);

            rig.Transition("t1", active: true);
            await WaitUntilAsync(() => collector.Has(onA, "t1") && collector.Has(onB, "t1"));

            rig.AttachedEventItems.ShouldBe(2);
            state.Apply(Cluster, "rev-2", [], [EquipmentGrant(EqB)], ScopePaths());
            rig.AttachedEventItems.ShouldBe(1);

            subscription.RemoveItem(onA);
            await subscription.ApplyChangesAsync(ct);
            onA.Status.Error.ShouldBeNull();

            // A restoring deployment must not re-attach the deleted item. The client could never see that
            // happen — it has deleted the item — so the server-side fan-out is what is asserted.
            state.Apply(Cluster, "rev-3", [], [EquipmentGrant(EqA), EquipmentGrant(EqB)], ScopePaths());
            rig.AttachedEventItems.ShouldBe(1, "the deleted item must not be resurrected onto EqA's fan-out");

            rig.Transition("t2", active: false);
            await WaitUntilAsync(() => collector.Has(onB, "t2"));
            await Task.Delay(TimeSpan.FromMilliseconds(700), ct);
            collector.Has(onA, "t2").ShouldBeFalse();

            await subscription.DeleteAsync(true, ct);
        }
        finally { await rig.DisposeAsync(); }
    }

    /// <summary>
    ///     The clock axis: the key behind a session's alarm subscription expires, one revalidation tick runs
    ///     through the host's own <see cref="AclRevalidationTimer"/>, and that session stops receiving alarms
    ///     — with no deployment anywhere in the run — while an independent session on the same notifier keeps
    ///     receiving the same event.
    /// </summary>
    [Fact]
    public async Task An_expired_key_stops_a_live_alarm_subscription_with_no_deploy()
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await Rig.StartAsync("AclEventExpiry", ct);
        try
        {
            var key = new ApiKey
            {
                ApiKeyId = KeyId,
                Name = KeyId,
                SecretHash = "hash",
                Principal = KeyPrincipal,
                Enabled = true,
                CreatedUtc = Start.AddDays(-1),
                ExpiresUtc = Start.AddMinutes(10),
            };
            var state = rig.Enforce(
                [key], ClusterGrant(KeyPrincipal, "acl-key"), ClusterGrant(OpcUaPrincipals.Anonymous, "acl-anon"));
            rig.Materialise();

            using var keySession = await rig.ConnectAsync(anonymous: false, ct);
            using var controlSession = await rig.ConnectAsync(anonymous: true, ct);

            var keyEvents = new EventCollector();
            var keySubscription = await CreateSubscriptionAsync(keySession, keyEvents, ct);
            var keyItem = AddEventItem(keySubscription, rig.UnsNodeId(keySession, EqA));
            await keySubscription.ApplyChangesAsync(ct);
            keyItem.Status.Error.ShouldBeNull();

            var controlEvents = new EventCollector();
            var controlSubscription = await CreateSubscriptionAsync(controlSession, controlEvents, ct);
            var controlItem = AddEventItem(controlSubscription, rig.UnsNodeId(controlSession, EqA));
            await controlSubscription.ApplyChangesAsync(ct);

            rig.Transition("t1", active: true);
            await WaitUntilAsync(() => keyEvents.Has(keyItem, "t1") && controlEvents.Has(controlItem, "t1"));

            // Only the clock moves.
            rig.Advance(TimeSpan.FromMinutes(20));
            rig.TickRevalidation();

            rig.Transition("t2", active: false);
            await WaitUntilAsync(() => controlEvents.Has(controlItem, "t2"));
            await Task.Delay(TimeSpan.FromMilliseconds(700), ct);
            keyEvents.Has(keyItem, "t2").ShouldBeFalse("the expired key's alarm feed must stop on the tick");
            state.Current.GenerationId.ShouldBe(1);

            await controlSubscription.DeleteAsync(true, ct);
            await keySubscription.DeleteAsync(true, ct);
        }
        finally { await rig.DisposeAsync(); }
    }

    /// <summary>
    ///     The negative control for the clock axis: the same run with the clock advanced but not past
    ///     <c>ExpiresUtc</c> keeps the alarm feed live across a tick.
    /// </summary>
    /// <remarks>
    ///     Without it the test above also passes on a build whose tick stands down every event item it sees,
    ///     which would silence every alarm consumer in the fleet once a minute.
    /// </remarks>
    [Fact]
    public async Task An_unexpired_key_keeps_its_alarm_subscription_across_a_tick()
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await Rig.StartAsync("AclEventLive", ct);
        try
        {
            var key = new ApiKey
            {
                ApiKeyId = KeyId,
                Name = KeyId,
                SecretHash = "hash",
                Principal = KeyPrincipal,
                Enabled = true,
                CreatedUtc = Start.AddDays(-1),
                ExpiresUtc = Start.AddHours(4),
            };
            rig.Enforce([key], ClusterGrant(KeyPrincipal, "acl-key"));
            rig.Materialise();

            using var session = await rig.ConnectAsync(anonymous: false, ct);
            var events = new EventCollector();
            var subscription = await CreateSubscriptionAsync(session, events, ct);
            var item = AddEventItem(subscription, rig.UnsNodeId(session, EqA));
            await subscription.ApplyChangesAsync(ct);

            rig.Transition("t1", active: true);
            await WaitUntilAsync(() => events.Has(item, "t1"));

            rig.Advance(TimeSpan.FromMinutes(20));
            rig.TickRevalidation();

            rig.Transition("t2", active: false);
            await WaitUntilAsync(() => events.Has(item, "t2"));

            await subscription.DeleteAsync(true, ct);
        }
        finally { await rig.DisposeAsync(); }
    }

    // ───────────────────────────── helpers ─────────────────────────────

    private static NodeAcl EquipmentGrant(string equipment) => new()
    {
        NodeAclId = $"acl-{equipment}",
        ClusterId = Cluster,
        Principal = OpcUaPrincipals.Anonymous,
        ScopeKind = NodeAclScopeKind.Equipment,
        ScopeId = $"eq-{equipment}",
        PermissionFlags = NodePermissions.Read | NodePermissions.Browse | NodePermissions.Subscribe,
    };

    private static NodeAcl ClusterGrant(string principal, string id) => new()
    {
        NodeAclId = id,
        ClusterId = Cluster,
        Principal = principal,
        ScopeKind = NodeAclScopeKind.Cluster,
        PermissionFlags = NodePermissions.Read | NodePermissions.Browse | NodePermissions.Subscribe,
    };

    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])],
        };

    private static async Task<Subscription> CreateSubscriptionAsync(
        ISession session, EventCollector collector, CancellationToken ct)
    {
        var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
        subscription.FastEventCallback = collector.OnEvents;
        session.AddSubscription(subscription);
        await subscription.CreateAsync(ct);
        return subscription;
    }

    private static MonitoredItem AddEventItem(Subscription subscription, NodeId source)
    {
        var filter = new EventFilter();
        filter.AddSelectClause(ObjectTypes.BaseEventType, BrowseNames.EventId);
        filter.AddSelectClause(ObjectTypes.BaseEventType, BrowseNames.Message);

        var item = new MonitoredItem(subscription.DefaultItem)
        {
            StartNodeId = source,
            AttributeId = Attributes.EventNotifier,
            Filter = filter,
            QueueSize = 100,
            SamplingInterval = 0,
        };
        subscription.AddItem(item);
        return item;
    }

    private 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");
    }

    /// <summary>Records which transition messages reached which monitored item.</summary>
    private sealed class EventCollector
    {
        private readonly object _gate = new();
        private readonly HashSet<(uint Handle, string Message)> _seen = [];

        public void OnEvents(Subscription subscription, EventNotificationList notification, IList<string> stringTable)
        {
            lock (_gate)
            {
                foreach (var e in notification.Events)
                {
                    if (e.EventFields.Count > MessageFieldIndex
                        && e.EventFields[MessageFieldIndex].Value is LocalizedText { Text: { } text })
                    {
                        _seen.Add((e.ClientHandle, text));
                    }
                }
            }
        }

        public bool Has(MonitoredItem item, string message)
        {
            lock (_gate) return _seen.Contains((item.ClientHandle, message));
        }
    }

    /// <summary>A booted server with ACL enforcement, one native alarm wired to two equipment notifiers.</summary>
    private sealed class Rig : IAsyncDisposable
    {
        private readonly string _pkiRoot;
        private readonly FakeClock _clock = new(Start);
        private OpcUaApplicationHost _host = null!;
        private OtOpcUaSdkServer _server = null!;
        private SdkAddressSpaceSink _sink = null!;
        private AclRevalidationTimer? _timer;
        private int _port;

        private Rig(string pkiRoot) => _pkiRoot = pkiRoot;

        private OtOpcUaNodeManager Nm => _server.NodeManager!;

        public static async Task<Rig> StartAsync(string name, CancellationToken ct)
        {
            var rig = new Rig(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 = [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>Switches enforcement on with the given grants and no keys.</summary>
        public DataPlaneAuthorityState Enforce(params NodeAcl[] grants) => Enforce([], grants);

        /// <summary>
        ///     Switches enforcement on, wiring the authorizer, the authority source (the deployment trigger)
        ///     and the host's revalidation timer (the clock trigger) over one hand-advanced clock.
        /// </summary>
        public DataPlaneAuthorityState Enforce(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: true, NullLogger.Instance, _clock);

            // A long interval: the tick is driven by hand through the same component the host arms.
            _timer = new AclRevalidationTimer(
                () => _server.NodeManager?.RevalidateMonitoredItems(),
                TimeSpan.FromHours(1),
                TimeProvider.System,
                NullLogger.Instance);
            return state;
        }

        /// <summary>One native condition on a raw device, wired as an event source of two equipment folders.</summary>
        public void Materialise()
        {
            _sink.EnsureFolder(RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
            _sink.MaterialiseAlarmCondition(
                RawAlarmPath, RawDeviceFolder, "temp_hi", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true);
            _sink.EnsureFolder(EqA, parentNodeId: null, displayName: "A", realm: AddressSpaceRealm.Uns);
            _sink.EnsureFolder(EqB, parentNodeId: null, displayName: "B", realm: AddressSpaceRealm.Uns);
            _sink.WireAlarmNotifiers(RawAlarmPath, AddressSpaceRealm.Raw, [EqA, EqB], AddressSpaceRealm.Uns);
        }

        /// <summary>Fires one genuine condition transition, carrying <paramref name="message"/>.</summary>
        public void Transition(string message, bool active) =>
            _sink.WriteAlarmCondition(
                RawAlarmPath,
                new AlarmConditionSnapshot(
                    Active: active, Acknowledged: !active, Confirmed: true, Enabled: true,
                    Shelving: AlarmShelvingKind.Unshelved, Severity: 700, Message: message),
                DateTime.UtcNow,
                AddressSpaceRealm.Raw);

        public NodeId UnsNodeId(ISession session, string id)
        {
            var ns = (ushort)session.NamespaceUris.GetIndex(V3NodeIds.UnsNamespaceUri);
            ns.ShouldBeGreaterThan((ushort)0);
            return new NodeId(id, ns);
        }

        /// <summary>Event items attached to the fan-out of the manager's own notifiers.</summary>
        public int AttachedEventItems => Nm.AttachedEventItemCount;

        public void Advance(TimeSpan by) => _clock.Advance(by);

        public void TickRevalidation() => _timer!.Tick();

        public async Task<ISession> ConnectAsync(bool anonymous, CancellationToken ct)
        {
            var appConfig = new ApplicationConfiguration
            {
                ApplicationName = "OtOpcUa.AclEventClient",
                ApplicationUri = $"urn:OtOpcUa.AclEventClient.{Guid.NewGuid():N}",
                ApplicationType = ApplicationType.Client,
                SecurityConfiguration = TestClientSecurity.Build(TestClientSecurity.AllocatePkiRoot()),
                ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 60_000 },
            };
            await appConfig.ValidateAsync(ApplicationType.Client, ct);
            appConfig.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true;

            var endpoint = await CoreClientUtils.SelectEndpointAsync(
                appConfig, $"opc.tcp://127.0.0.1:{_port}/OtOpcUa", false, DefaultTelemetry.Create(_ => { }), ct);

            var identity = anonymous
                ? new UserIdentity(new AnonymousIdentityToken())
                : new UserIdentity(new UserNameIdentityToken
                {
                    UserName = "apikey",
                    // 1.5.378 carries the UserName token password as raw bytes; the host decodes UTF-8.
                    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: "AclEventSubscriptionEnforcementTests",
                sessionTimeout: 60_000,
                identity: identity,
                preferredLocales: null,
                ct: ct);
        }

        public async ValueTask DisposeAsync()
        {
            _timer?.Dispose();
            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 the fixed <c>apikey</c> username as <see cref="KeyPrincipal"/> carrying
        ///     <see cref="KeyId"/>; anonymous falls through to the interface default (the
        ///     <c>anonymous</c> principal). Credential verification itself is covered in the Host suite.
        /// </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, [], KeyId)
                        : OpcUaUserAuthResult.Deny("Invalid API key."));
        }

        /// <summary>A hand-advanced clock; the authorizer reads it on every decision.</summary>
        private sealed class FakeClock(DateTime utcNow) : TimeProvider
        {
            private DateTimeOffset _now = new(utcNow, TimeSpan.Zero);

            /// <inheritdoc />
            public override DateTimeOffset GetUtcNow() => _now;

            public void Advance(TimeSpan by) => _now = _now.Add(by);
        }
    }
}