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

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