lmxopcua/src/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host/Ipc/PipeServer.cs

using System;
using System.IO.Pipes;
using System.Security.Principal;
using System.Threading;
using System.Threading.Tasks;
using MessagePack;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Contracts;

namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host.Ipc;

/// <summary>
///     Accepts one client connection at a time on a named pipe with the strict ACL from
///     <see cref="PipeAcl"/>. Verifies the peer SID and the per-process shared secret before any
///     RPC frame is accepted. Per <c>driver-stability.md §"IPC Security"</c>.
/// </summary>
public sealed class PipeServer : IDisposable
{
    private readonly string _pipeName;
    private readonly SecurityIdentifier _allowedSid;
    private readonly string _sharedSecret;
    private readonly ILogger _logger;
    private readonly CancellationTokenSource _cts = new();
    private NamedPipeServerStream? _current;

    public PipeServer(string pipeName, SecurityIdentifier allowedSid, string sharedSecret, ILogger logger)
    {
        _pipeName = pipeName ?? throw new ArgumentNullException(nameof(pipeName));
        _allowedSid = allowedSid ?? throw new ArgumentNullException(nameof(allowedSid));
        _sharedSecret = sharedSecret ?? throw new ArgumentNullException(nameof(sharedSecret));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
    }

    /// <summary>
    ///     Accepts one connection, performs Hello handshake, then dispatches frames to
    ///     <paramref name="handler"/> until EOF or cancel. Returns when the client disconnects.
    /// </summary>
    public async Task RunOneConnectionAsync(IFrameHandler handler, CancellationToken ct)
    {
        using var linked = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token, ct);
        var acl = PipeAcl.Create(_allowedSid);

        // .NET Framework 4.8 uses the legacy constructor overload that takes a PipeSecurity directly.
        _current = new NamedPipeServerStream(
            _pipeName,
            PipeDirection.InOut,
            maxNumberOfServerInstances: 1,
            PipeTransmissionMode.Byte,
            PipeOptions.Asynchronous,
            inBufferSize: 64 * 1024,
            outBufferSize: 64 * 1024,
            pipeSecurity: acl);

        try
        {
            await _current.WaitForConnectionAsync(linked.Token).ConfigureAwait(false);

            if (!VerifyCaller(_current, out var reason))
            {
                _logger.Warning("IPC caller rejected: {Reason}", reason);
                _current.Disconnect();
                return;
            }

            using var reader = new FrameReader(_current, leaveOpen: true);
            using var writer = new FrameWriter(_current, leaveOpen: true);

            // First frame must be a Hello with the correct shared secret.
            var first = await reader.ReadFrameAsync(linked.Token).ConfigureAwait(false);
            if (first is null || first.Value.Kind != MessageKind.Hello)
            {
                _logger.Warning("IPC first frame was not Hello; dropping");
                return;
            }

            var hello = MessagePackSerializer.Deserialize<Hello>(first.Value.Body);
            if (!string.Equals(hello.SharedSecret, _sharedSecret, StringComparison.Ordinal))
            {
                await writer.WriteAsync(MessageKind.HelloAck,
                    new HelloAck { Accepted = false, RejectReason = "shared-secret-mismatch" },
                    linked.Token).ConfigureAwait(false);
                _logger.Warning("IPC Hello rejected: shared-secret-mismatch");
                return;
            }

            if (hello.ProtocolMajor != Hello.CurrentMajor)
            {
                await writer.WriteAsync(MessageKind.HelloAck,
                    new HelloAck { Accepted = false, RejectReason = $"major-version-mismatch-peer={hello.ProtocolMajor}-server={Hello.CurrentMajor}" },
                    linked.Token).ConfigureAwait(false);
                _logger.Warning("IPC Hello rejected: major mismatch peer={Peer} server={Server}",
                    hello.ProtocolMajor, Hello.CurrentMajor);
                return;
            }

            await writer.WriteAsync(MessageKind.HelloAck,
                new HelloAck { Accepted = true, HostName = Environment.MachineName },
                linked.Token).ConfigureAwait(false);

            using var attachment = handler.AttachConnection(writer);

            while (!linked.Token.IsCancellationRequested)
            {
                var frame = await reader.ReadFrameAsync(linked.Token).ConfigureAwait(false);
                if (frame is null) break;

                await handler.HandleAsync(frame.Value.Kind, frame.Value.Body, writer, linked.Token).ConfigureAwait(false);
            }
        }
        finally
        {
            _current.Dispose();
            _current = null;
        }
    }

    /// <summary>
    ///     Runs the server continuously, handling one connection at a time. When a connection ends
    ///     (clean or error), accepts the next.
    /// </summary>
    public async Task RunAsync(IFrameHandler handler, CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            try { await RunOneConnectionAsync(handler, ct).ConfigureAwait(false); }
            catch (OperationCanceledException) { break; }
            catch (Exception ex) { _logger.Error(ex, "IPC connection loop error — accepting next"); }
        }
    }

    private bool VerifyCaller(NamedPipeServerStream pipe, out string reason)
    {
        try
        {
            pipe.RunAsClient(() =>
            {
                using var wi = WindowsIdentity.GetCurrent();
                if (wi.User is null)
                    throw new InvalidOperationException("GetCurrent().User is null — cannot verify caller");
                if (wi.User != _allowedSid)
                    throw new UnauthorizedAccessException(
                        $"caller SID {wi.User.Value} does not match allowed {_allowedSid.Value}");
            });
            reason = string.Empty;
            return true;
        }
        catch (Exception ex) { reason = ex.Message; return false; }
    }

    public void Dispose()
    {
        _cts.Cancel();
        _current?.Dispose();
        _cts.Dispose();
    }
}

public interface IFrameHandler
{
    Task HandleAsync(MessageKind kind, byte[] body, FrameWriter writer, CancellationToken ct);

    /// <summary>
    ///     Called once per accepted connection after the Hello handshake. Lets the handler
    ///     attach server-pushed event sinks (data-change, alarm, host-status) to the
    ///     connection's <paramref name="writer"/>. Returns an <see cref="IDisposable"/> the
    ///     pipe server disposes when the connection closes — backends use it to unsubscribe.
    ///     Implementations that don't push events can return <see cref="NoopAttachment"/>.
    /// </summary>
    IDisposable AttachConnection(FrameWriter writer);

    public sealed class NoopAttachment : IDisposable
    {
        public static readonly NoopAttachment Instance = new();
        public void Dispose() { }
    }
}