lmxopcua/src/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.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.FOCAS.Shared;
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Shared.Contracts;

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

/// <summary>
///     Accepts one client connection at a time on the FOCAS Host's named pipe with the
///     strict ACL from <see cref="PipeAcl"/>. Verifies the peer SID + per-process shared
///     secret before any RPC frame is accepted. Mirrors the Galaxy.Host pipe server byte for
///     byte — different MessageKind enum, same negotiation semantics.
/// </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));
    }

    public async Task RunOneConnectionAsync(IFrameHandler handler, CancellationToken ct)
    {
        using var linked = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token, ct);
        var acl = PipeAcl.Create(_allowedSid);

        _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("FOCAS IPC caller rejected: {Reason}", reason);
                _current.Disconnect();
                return;
            }

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

            var first = await reader.ReadFrameAsync(linked.Token).ConfigureAwait(false);
            if (first is null || first.Value.Kind != FocasMessageKind.Hello)
            {
                _logger.Warning("FOCAS 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(FocasMessageKind.HelloAck,
                    new HelloAck { Accepted = false, RejectReason = "shared-secret-mismatch" },
                    linked.Token).ConfigureAwait(false);
                _logger.Warning("FOCAS IPC Hello rejected: shared-secret-mismatch");
                return;
            }

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

            await writer.WriteAsync(FocasMessageKind.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;
        }
    }

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