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

using System.Net;
using System.Net.Sockets;
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.OpcUaServer.Security;

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

/// <summary>
///     #520 stage 6 over the wire — browse filtering, subscription denial, and the one that could not be
///     settled from documentation: whether revoking a grant actually stops a <b>live</b> subscription.
/// </summary>
/// <remarks>
///     <para>
///         These are here rather than in the unit project because the mechanism under test lives in the SDK.
///         <c>RevalidateMonitoredItems</c> re-authorizes each item using
///         <c>IMonitoredItem.EffectiveIdentity</c> — the identity of the session that created it — and then
///         disables the item. Both halves are assumptions about SDK behaviour, and both fail silently in the
///         dangerous direction if wrong: an unrecoverable identity denies every item, which would disable
///         every subscription on the fleet the first time anyone deployed with the flag on.
///     </para>
///     <para>
///         The client connects <b>anonymously</b>, which under #520's D3 is a normal grantable principal
///         (<c>anonymous</c>) rather than a bypass — so these tests exercise the real grant path with no
///         key-minting ceremony.
///     </para>
///     <para><b>Run note:</b> heavy in-process server+client tests; part of the serial integration pass.</para>
/// </remarks>
public sealed class AclSubscriptionEnforcementTests
{
    private const string Cluster = "SITE-A";
    private const string Device = "Plant/Modbus/dev1";
    private const string Tag = "Plant/Modbus/dev1/Speed";

    /// <summary>
    ///     THE test this file exists for: a deployment that removes the grant stops an already-running
    ///     subscription, and tells the client why.
    /// </summary>
    /// <remarks>
    ///     Without it, revocation would close the operations nobody uses (a one-shot read, an occasional
    ///     write) and leave the streaming feed — which is how ScadaBridge and every other machine consumer
    ///     actually reads this server — running indefinitely on a grant that no longer exists.
    /// </remarks>
    [Fact]
    public async Task Revoking_a_grant_stops_a_live_subscription()
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await Rig.StartAsync("AclRevoke", ct);
        try
        {
            var state = rig.Enforce(Grant(NodePermissions.Read | NodePermissions.Browse | NodePermissions.Subscribe));
            rig.Materialise();
            rig.Write(1);

            using var session = await rig.ConnectAsync(ct);
            var (subscription, received) = await rig.SubscribeAsync(session, ct);

            // Baseline: the feed is live.
            await WaitUntilAsync(() => received.Any(v => Equals(v.Value, 1f)), TimeSpan.FromSeconds(10));

            received.Clear();

            // The next deployment carries no grants at all.
            state.Apply(Cluster, "rev-2", [], [], NoScopes);

            // The client is TOLD, rather than watching a feed go quiet for no stated reason.
            await WaitUntilAsync(
                () => received.Any(v => v.StatusCode.Code == StatusCodes.BadUserAccessDenied),
                TimeSpan.FromSeconds(10));

            // And the data really has stopped: a fresh value never arrives.
            received.Clear();
            rig.Write(2);
            await Task.Delay(TimeSpan.FromSeconds(2), ct);
            received.ShouldNotContain(v => Equals(v.Value, 2f));

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

    /// <summary>
    ///     The positive control for the test above, and the one that would catch the failure mode that
    ///     worries me most: if the identity behind a monitored item could not be recovered, every item would
    ///     be denied and the fleet's subscriptions would all stop on the first deploy. Here a re-deploy that
    ///     KEEPS the grant must leave the feed running.
    /// </summary>
    [Fact]
    public async Task A_redeploy_that_keeps_the_grant_leaves_the_subscription_running()
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await Rig.StartAsync("AclKeep", ct);
        try
        {
            var grant = Grant(NodePermissions.Read | NodePermissions.Browse | NodePermissions.Subscribe);
            var state = rig.Enforce(grant);
            rig.Materialise();
            rig.Write(1);

            using var session = await rig.ConnectAsync(ct);
            var (subscription, received) = await rig.SubscribeAsync(session, ct);
            await WaitUntilAsync(() => received.Any(v => Equals(v.Value, 1f)), TimeSpan.FromSeconds(10));

            received.Clear();
            state.Apply(Cluster, "rev-2", [], [grant], NoScopes);

            rig.Write(2);
            await WaitUntilAsync(() => received.Any(v => Equals(v.Value, 2f)), TimeSpan.FromSeconds(10));
            received.ShouldNotContain(v => v.StatusCode.Code == StatusCodes.BadUserAccessDenied);

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

    /// <summary>
    ///     A grant restored by a later deployment brings the feed back on its own, without the client
    ///     doing anything.
    /// </summary>
    /// <remarks>
    ///     This is why a denied item is detached from its node's fan-out rather than set
    ///     <c>MonitoringMode.Disabled</c>. A disabled item can only be re-enabled by the client, and a
    ///     machine consumer has no reason to ever try — so an operator who revoked a grant by mistake and
    ///     put it straight back would have left the subscription dead until someone restarted the consumer.
    /// </remarks>
    [Fact]
    public async Task Restoring_a_grant_brings_the_subscription_back()
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await Rig.StartAsync("AclRestore", ct);
        try
        {
            var grant = Grant(NodePermissions.Read | NodePermissions.Browse | NodePermissions.Subscribe);
            var state = rig.Enforce(grant);
            rig.Materialise();
            rig.Write(1);

            using var session = await rig.ConnectAsync(ct);
            var (subscription, received) = await rig.SubscribeAsync(session, ct);
            await WaitUntilAsync(() => received.Any(v => Equals(v.Value, 1f)), TimeSpan.FromSeconds(10));

            state.Apply(Cluster, "rev-2", [], [], NoScopes);
            await WaitUntilAsync(
                () => received.Any(v => v.StatusCode.Code == StatusCodes.BadUserAccessDenied),
                TimeSpan.FromSeconds(10));

            // The operator puts the grant back.
            received.Clear();
            state.Apply(Cluster, "rev-3", [], [grant], NoScopes);

            rig.Write(3);
            await WaitUntilAsync(() => received.Any(v => Equals(v.Value, 3f)), TimeSpan.FromSeconds(10));

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

    /// <summary>A principal with no Subscribe bit cannot create a monitored item at all.</summary>
    [Fact]
    public async Task Subscribing_without_the_permission_is_refused()
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await Rig.StartAsync("AclNoSub", ct);
        try
        {
            // Read and Browse, deliberately NOT Subscribe — so a failure here cannot be "the session was
            // denied everything".
            rig.Enforce(Grant(NodePermissions.Read | NodePermissions.Browse));
            rig.Materialise();
            rig.Write(1);

            using var session = await rig.ConnectAsync(ct);

            // The read works, proving the session is admitted and granted.
            var read = await session.ReadValueAsync(rig.TagNodeId(session), ct);
            read.StatusCode.Code.ShouldNotBe(StatusCodes.BadUserAccessDenied);

            var (subscription, _) = await rig.SubscribeAsync(session, ct, expectFailure: true);
            subscription.MonitoredItems.First().Status.Error?.StatusCode.Code
                .ShouldBe(StatusCodes.BadUserAccessDenied);

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

    /// <summary>
    ///     Browse returns only what the principal may see: a grant on one device makes that device visible
    ///     under its parent folder and its sibling invisible.
    /// </summary>
    /// <remarks>
    ///     The parent folder itself must remain browsable — that is the ancestor-visibility ("transit")
    ///     rule. If it were not, the grant on the device below would be unreachable and the operator's rule
    ///     would silently do nothing.
    /// </remarks>
    [Fact]
    public async Task Browse_returns_only_the_granted_subtree()
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await Rig.StartAsync("AclBrowse", ct);
        try
        {
            var scoped = new NodeAcl
            {
                NodeAclId = "acl-dev1",
                ClusterId = Cluster,
                Principal = OpcUaPrincipals.Anonymous,
                ScopeKind = NodeAclScopeKind.RawDevice,
                ScopeId = "dev-1",
                PermissionFlags = NodePermissions.Browse | NodePermissions.Read,
            };
            var scopes = new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>
            {
                [new AclScopeRef(NodeAclScopeKind.RawDevice, "dev-1")] =
                    [new NodeAclPath([NodeAclRealms.Raw, "Plant", "Modbus", "dev1"])],
            };

            rig.Enforce(scoped, scopes);
            rig.MaterialiseTwoDevices();

            using var session = await rig.ConnectAsync(ct);
            var ns = rig.RawNamespace(session);

            var children = await BrowseChildrenAsync(session, new NodeId("Plant/Modbus", ns), ct);

            children.ShouldContain(n => n == "Plant/Modbus/dev1");
            children.ShouldNotContain(n => n == "Plant/Modbus/dev9");
        }
        finally { await rig.DisposeAsync(); }
    }

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

    private static readonly IReadOnlyDictionary<AclScopeRef, IReadOnlyList<NodeAclPath>> NoScopes =
        new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>();

    private static NodeAcl Grant(NodePermissions flags) => new()
    {
        NodeAclId = "acl-cluster",
        ClusterId = Cluster,
        Principal = OpcUaPrincipals.Anonymous,
        ScopeKind = NodeAclScopeKind.Cluster,
        PermissionFlags = flags,
    };

    private static async Task<List<string>> BrowseChildrenAsync(
        ISession session, NodeId parent, CancellationToken ct)
    {
        var response = await session.BrowseAsync(
            requestHeader: null,
            view: null,
            requestedMaxReferencesPerNode: 0,
            nodesToBrowse:
            [
                new BrowseDescription
                {
                    NodeId = parent,
                    BrowseDirection = BrowseDirection.Forward,
                    ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences,
                    IncludeSubtypes = true,
                    NodeClassMask = 0,
                    ResultMask = (uint)BrowseResultMask.All,
                },
            ],
            ct: ct);

        return response.Results[0].References
            .Select(r => r.NodeId.Identifier as string ?? string.Empty)
            .ToList();
    }

    private static async Task WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
    {
        var deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            if (condition()) return;
            await Task.Delay(50);
        }
        condition().ShouldBeTrue("condition not met within timeout");
    }

    /// <summary>A booted server plus the ACL wiring, so each test states only what it is actually about.</summary>
    private sealed class Rig : IAsyncDisposable
    {
        private readonly string _pkiRoot;
        private OpcUaApplicationHost _host = null!;
        private OtOpcUaSdkServer _server = null!;
        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);

            await rig._host.StartAsync(rig._server, ct);
            return rig;
        }

        /// <summary>Switches enforcement on with one grant, returning the state so a test can redeploy.</summary>
        public DataPlaneAuthorityState Enforce(
            NodeAcl grant, IReadOnlyDictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>? scopes = null)
        {
            var state = new DataPlaneAuthorityState();
            state.Apply(Cluster, "rev-1", [], [grant], scopes ?? NoScopes);

            // BOTH must be set: Authorizer decides, and DataPlaneAuthority is what carries the Changed
            // signal that reaches live subscriptions. Wiring only the first passes every test that makes a
            // request and none that revokes.
            Nm.DataPlaneAuthority = state;
            Nm.Authorizer = new DataPlaneAuthorizer(
                state, new TriePermissionEvaluator(state.Tries), enabled: true, NullLogger.Instance);
            return state;
        }

        public void Materialise()
        {
            Nm.EnsureFolder("Plant", null, "Plant", AddressSpaceRealm.Raw);
            Nm.EnsureFolder("Plant/Modbus", "Plant", "Modbus", AddressSpaceRealm.Raw);
            Nm.EnsureFolder(Device, "Plant/Modbus", "dev1", AddressSpaceRealm.Raw);
            Nm.EnsureVariable(Tag, Device, "Speed", "Float", writable: false, AddressSpaceRealm.Raw);
        }

        public void MaterialiseTwoDevices()
        {
            Materialise();
            Nm.EnsureFolder("Plant/Modbus/dev9", "Plant/Modbus", "dev9", AddressSpaceRealm.Raw);
            Nm.EnsureVariable(
                "Plant/Modbus/dev9/Torque", "Plant/Modbus/dev9", "Torque", "Float",
                writable: false, AddressSpaceRealm.Raw);
        }

        public void Write(float value) =>
            Nm.WriteValue(Tag, value, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);

        public ushort RawNamespace(ISession session)
        {
            var ns = (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri);
            ns.ShouldBeGreaterThan((ushort)0);
            return ns;
        }

        public NodeId TagNodeId(ISession session) => new(Tag, RawNamespace(session));

        public async Task<(Subscription Subscription, List<DataValue> Received)> SubscribeAsync(
            ISession session, CancellationToken ct, bool expectFailure = false)
        {
            var received = new List<DataValue>();
            var gate = new object();

            var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
            session.AddSubscription(subscription);
            await subscription.CreateAsync(ct);

            var item = new MonitoredItem(subscription.DefaultItem)
            {
                StartNodeId = TagNodeId(session),
                AttributeId = Attributes.Value,
                SamplingInterval = 50,
                QueueSize = 10,
            };
            item.Notification += (mi, _) =>
            {
                foreach (var v in mi.DequeueValues()) lock (gate) received.Add(v);
            };
            subscription.AddItem(item);
            await subscription.ApplyChangesAsync(ct);

            if (!expectFailure) item.Status.Error.ShouldBeNull();

            // The list is only ever appended under `gate` and read from the test thread between awaits;
            // a snapshot-on-read wrapper would hide ordering rather than help.
            return (subscription, received);
        }

        public async Task<ISession> ConnectAsync(CancellationToken ct)
        {
            var appConfig = new ApplicationConfiguration
            {
                ApplicationName = "OtOpcUa.AclClient",
                ApplicationUri = $"urn:OtOpcUa.AclClient.{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);

            return await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(
                appConfig,
                new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(appConfig)),
                updateBeforeConnect: false,
                checkDomain: false,
                sessionName: "AclSubscriptionEnforcementTests",
                sessionTimeout: 60_000,
                identity: new UserIdentity(new AnonymousIdentityToken()),
                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;
        }
    }
}