diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 2643b862..39db2252 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -39,6 +39,7 @@ services: ports: - "9001:5000" # Web UI + Inbound API - "9011:8081" # Akka remoting (host access for CLI/debugging) + - "9013:8083" # gRPC control plane (CentralControlService, T1A.2) volumes: - ./central-node-a/appsettings.Central.json:/app/appsettings.Central.json:ro - ./central-node-a/logs:/app/logs @@ -86,6 +87,7 @@ services: ports: - "9002:5000" # Web UI + Inbound API - "9012:8081" # Akka remoting + - "9014:8083" # gRPC control plane (CentralControlService, T1A.2) volumes: - ./central-node-b/appsettings.Central.json:/app/appsettings.Central.json:ro - ./central-node-b/logs:/app/logs diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs new file mode 100644 index 00000000..5ffcba68 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs @@ -0,0 +1,299 @@ +using Akka.Actor; +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification; +using GrpcStatus = Grpc.Core.Status; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +/// +/// Central-hosted gRPC face of the seven site→central control messages +/// (Protos/central_control.proto). Decodes each request onto the SAME in-process +/// message type the Akka ClusterClient path already carries, Asks +/// , and encodes the reply back. +/// +/// +/// +/// Direction is inverted from . That server runs on a +/// site and central dials in; this one runs on CENTRAL and the site dials in. The two listen on +/// the same port number (8083) on their respective nodes, which is symmetry, not a collision — +/// a node is either central or a site, never both. +/// +/// +/// Zero handler logic lives here. Every RPC lands on a receive +/// CentralCommunicationActor already implements for the ClusterClient path, so the two +/// transports cannot drift in behaviour: the actor is the single implementation, and this class +/// is a codec plus an Ask. That is also why the service takes the actor through +/// rather than resolving anything from DI — the actor is created by the +/// host's Akka bootstrap, not by the container. +/// +/// +/// Fault semantics deliberately differ from 's ingest +/// RPCs. That server answers a failed audit ingest with an EMPTY IngestAck; this one +/// fails the call with a non-OK status. Both leave the site's rows Pending for the next +/// drain, so the outcome is the same — but this service replaces the ClusterClient path, whose +/// documented behaviour is to propagate the fault (CentralCommunicationActor's +/// HandleIngestAuditEvents pipes a Status.Failure back), and preserving that keeps +/// a lost batch visible as a failure rather than as a successful call that acked nothing. +/// +/// +/// Status mapping, and why it is not uniform. A site transport may safely re-send a call +/// to the peer central node only when the call provably never ran. So: +/// +/// +/// — this node is not ready; nothing was dispatched, +/// so a cross-node retry is safe and correct. +/// — the Ask timed out. The message WAS +/// delivered and may have been processed; retrying it on the other node would duplicate work. +/// — the handler faulted (a piped +/// , e.g. a database error inside reconcile). Same +/// reasoning: it ran, so do not re-send it elsewhere. +/// +/// +public sealed class CentralControlGrpcService : CentralControlService.CentralControlServiceBase +{ + private readonly ILogger _logger; + private readonly CommunicationOptions _options; + + // Null until the host's Akka bootstrap hands the actor over. Doubles as the readiness + // flag: a call arriving before then cannot be served and is refused with Unavailable. + // Volatile because SetReady runs on the startup thread while calls are served on + // Kestrel's thread pool. + private volatile IActorRef? _central; + + /// + /// Creates the service. This must remain the only public constructor — see + /// for how the actor arrives, and the Host's + /// CentralControlAuthInterceptor for the interceptor-side version of the same rule. + /// + /// Logger for readiness and fault diagnostics. + /// Communication options supplying the per-RPC Ask timeouts. + public CentralControlGrpcService( + ILogger logger, + IOptions options) + { + ArgumentNullException.ThrowIfNull(logger); + ArgumentNullException.ThrowIfNull(options); + + _logger = logger; + _options = options.Value; + } + + /// + /// Hands the CentralCommunicationActor to the service and opens the gate. Mirrors + /// : the gRPC service is a DI singleton created + /// before the actor system exists, so the actor arrives post-construction. + /// + /// + /// The contract is deliberately narrow, exactly as on the site side: it asserts that the + /// actor exists and can receive, NOT that every downstream singleton proxy + /// (notification-outbox, audit-log-ingest) has registered itself yet. Those + /// register moments later in the same startup path, and the actor already answers a call + /// that beats them with the same "not available, retry" reply it gives on the ClusterClient + /// path — so gating readiness on them would add nothing but a longer window in which sites + /// see . + /// + /// The central communication actor. + public void SetReady(IActorRef centralCommunicationActor) + { + ArgumentNullException.ThrowIfNull(centralCommunicationActor); + _central = centralCommunicationActor; + } + + /// Exposed for wiring assertions in tests. + internal bool IsReady => _central is not null; + + /// + public override async Task SubmitNotification( + NotificationSubmitDto request, ServerCallContext context) + { + var central = RequireReady(context); + var ack = await AskAsync( + central, + CentralControlDtoMapper.FromDto(request), + _options.NotificationForwardTimeout, + context).ConfigureAwait(false); + + return CentralControlDtoMapper.ToDto(ack); + } + + /// + public override async Task QueryNotificationStatus( + NotificationStatusQueryDto request, ServerCallContext context) + { + var central = RequireReady(context); + var response = await AskAsync( + central, + CentralControlDtoMapper.FromDto(request), + _options.NotificationForwardTimeout, + context).ConfigureAwait(false); + + return CentralControlDtoMapper.ToDto(response); + } + + /// + public override async Task IngestAuditEvents( + AuditEventBatch request, ServerCallContext context) + { + // An empty batch is a no-op the actor need never see; answering it here also means a + // site that drains an empty queue does not fail against a not-yet-ready central. + if (request.Events.Count == 0) + { + return new IngestAck(); + } + + var central = RequireReady(context); + var reply = await AskAsync( + central, + CentralControlDtoMapper.FromDto(request), + SiteStreamGrpcServer.AuditIngestAskTimeout, + context).ConfigureAwait(false); + + return CentralControlDtoMapper.ToIngestAck(reply.AcceptedEventIds); + } + + /// + public override async Task IngestCachedTelemetry( + CachedTelemetryBatch request, ServerCallContext context) + { + if (request.Packets.Count == 0) + { + return new IngestAck(); + } + + var central = RequireReady(context); + var reply = await AskAsync( + central, + CentralControlDtoMapper.FromDto(request), + SiteStreamGrpcServer.AuditIngestAskTimeout, + context).ConfigureAwait(false); + + return CentralControlDtoMapper.ToIngestAck(reply.AcceptedEventIds); + } + + /// + public override async Task ReconcileSite( + ReconcileSiteRequestDto request, ServerCallContext context) + { + var central = RequireReady(context); + var response = await AskAsync( + central, + CentralControlDtoMapper.FromDto(request), + _options.QueryTimeout, + context).ConfigureAwait(false); + + return CentralControlDtoMapper.ToDto(response); + } + + /// + public override async Task ReportSiteHealth( + SiteHealthReportDto request, ServerCallContext context) + { + var central = RequireReady(context); + var ack = await AskAsync( + central, + CentralControlDtoMapper.FromDto(request), + _options.HealthReportTimeout, + context).ConfigureAwait(false); + + return CentralControlDtoMapper.ToDto(ack); + } + + /// + /// Application heartbeat — always answers OK, even when this node is not ready. + /// + /// + /// The heartbeat is fire-and-forget on both sides: nothing on the site consumes the reply, + /// and the site's heartbeat timer must never take a fault (a failing heartbeat that raised + /// an error would be a self-inflicted outage on a purely informational signal). So this is + /// the one RPC that does not go through : a heartbeat arriving + /// before the actor exists is logged and dropped, exactly as the actor itself drops one + /// that arrives before ICentralHealthAggregator is resolvable. Liveness is still + /// detected — the aggregator's offline timeout fires when the heartbeats stop landing. + /// + /// The heartbeat. + /// The gRPC call context. + /// An empty reply, always. + public override Task Heartbeat(HeartbeatDto request, ServerCallContext context) + { + var central = _central; + if (central is null) + { + _logger.LogDebug( + "Dropped a heartbeat from site {SiteId}: the central communication actor is not " + + "ready yet. Heartbeats are fire-and-forget, so the call still succeeds.", + request.SiteId); + return Task.FromResult(new Empty()); + } + + // Tell, never Ask: the actor's HandleHeartbeat sends no reply. + central.Tell(CentralControlDtoMapper.FromDto(request), ActorRefs.NoSender); + return Task.FromResult(new Empty()); + } + + /// + /// Returns the central communication actor, or throws + /// when the host has not finished bringing the actor system up. + /// + private IActorRef RequireReady(ServerCallContext context) + { + var central = _central; + if (central is not null) + { + return central; + } + + _logger.LogWarning( + "Refused a control-plane call to {Method}: the central communication actor is not " + + "ready yet. Nothing was dispatched, so the caller may retry (including against " + + "the peer central node).", + context.Method); + + throw new RpcException(new GrpcStatus( + StatusCode.Unavailable, + "Central control plane is not ready: the actor system is still starting.")); + } + + /// + /// Asks the central actor and maps a fault onto the status code that tells the caller + /// whether a cross-node retry is safe. See the class remarks for the mapping rationale. + /// + private async Task AskAsync( + IActorRef central, object message, TimeSpan timeout, ServerCallContext context) + { + try + { + return await central.Ask(message, timeout, context.CancellationToken) + .ConfigureAwait(false); + } + catch (AskTimeoutException ex) + { + _logger.LogWarning(ex, + "Control-plane call {Method} timed out after {Timeout} waiting for the central " + + "communication actor.", + context.Method, timeout); + throw new RpcException(new GrpcStatus( + StatusCode.DeadlineExceeded, + $"Central did not answer within {timeout}.")); + } + catch (OperationCanceledException) + { + // The client gave up or its deadline expired; there is no one left to answer. + throw new RpcException(new GrpcStatus( + StatusCode.Cancelled, "The call was cancelled.")); + } + catch (Exception ex) + { + _logger.LogError(ex, + "Control-plane call {Method} faulted inside the central communication actor.", + context.Method); + throw new RpcException(new GrpcStatus( + StatusCode.Internal, "Central failed to process the call.")); + } + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index fb41a772..7d58cc63 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -436,6 +436,19 @@ akka {{ ClusterClientReceptionist.Get(_actorSystem).RegisterService(centralCommActor); _logger.LogInformation("CentralCommunicationActor registered with ClusterClientReceptionist"); + // Hand the same actor to the central-hosted gRPC control plane (T1A.2) and open its + // readiness gate — the gRPC face Asks this exact actor, so both transports resolve to + // one handler implementation. Mirrors SiteStreamGrpcServer.SetReady on the site side: + // the service is a DI singleton created before the actor system exists, so the actor + // arrives here post-construction. Null on a host that did not register the service + // (e.g. an in-process test harness), so the wiring is a guarded no-op there. + var centralControlGrpc = _serviceProvider + .GetService(); + centralControlGrpc?.SetReady(centralCommActor); + _logger.LogInformation( + "CentralControlGrpcService readiness set (service bound: {Bound})", + centralControlGrpc is not null); + // Wire up the CommunicationService with the actor reference var commService = _serviceProvider.GetService(); commService?.SetCommunicationActor(centralCommActor); diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/CentralControlAuthInterceptor.cs b/src/ZB.MOM.WW.ScadaBridge.Host/CentralControlAuthInterceptor.cs new file mode 100644 index 00000000..9fb71bb0 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Host/CentralControlAuthInterceptor.cs @@ -0,0 +1,245 @@ +using System.Security.Cryptography; +using System.Text; +using Grpc.Core; +using Grpc.Core.Interceptors; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Host; + +/// +/// Gates the central-hosted gRPC control plane (CentralControlService) with each site's +/// preshared key. The sibling of , but with the +/// verification model inverted: a site checks one bearer token against its own single key, whereas +/// central must check the presented token against the key belonging to the SITE that sent it. +/// +/// +/// +/// Why a separate class rather than a second constructor on +/// . Grpc.AspNetCore registers a +/// type-registered interceptor through InterceptorRegistration.GetFactory(), which throws +/// "Multiple constructors accepting all given argument types have been found" the moment a +/// second public constructor is applicable. That throw lands inside the pipeline on every call and +/// surfaces as Unknown / "Exception was thrown by handler" — the node boots healthy and +/// every gated call dies looking like a handler bug, with correct/wrong/no key all producing the +/// identical error. It shipped once with a fully green suite and was caught only on the rig. +/// Central's model genuinely differs (per-site key by header, not the site's one own-key), so it +/// gets its own class with its own single public constructor rather than a variant ctor on the +/// site interceptor. Both classes are pinned by a reflection test asserting exactly one public +/// constructor. +/// +/// +/// Fail-closed on every branch. A gated call is refused with +/// when: the x-scadabridge-site header is +/// missing or blank; no key can be resolved for that site ( throws); +/// or the presented bearer token does not match. There is no pass-through — an unresolvable or +/// absent identity never degrades to "let it in". Non-gated services (should any share the +/// listener) return immediately, matching the site interceptor's shape. +/// +/// +/// The comparison is over UTF-8, the same +/// constant-time compare the site interceptor and LocalDb sync use. +/// +/// +public sealed class CentralControlAuthInterceptor : Interceptor +{ + /// + /// Service prefixes gated by default — the one central-hosted control-plane service. Taken + /// from the generated package scadabridge.centralcontrol.v1; service CentralControlService. + /// + public static readonly IReadOnlyList DefaultGatedPrefixes = + new[] { $"/{CentralControlService.Descriptor.FullName}/" }; + + private readonly IReadOnlyList _gatedPrefixes; + private readonly ISitePskProvider _pskProvider; + private readonly ILogger _logger; + + /// + /// Creates the interceptor gating . + /// + /// + /// This must remain the ONLY public constructor — see the class remarks for why a + /// second one silently disables the gate. Pinned by + /// CentralControlAuthInterceptorTests.TheInterceptorHasExactlyOnePublicConstructor. + /// + /// Resolves each site's preshared key. + /// Logger for denial diagnostics. + public CentralControlAuthInterceptor( + ISitePskProvider pskProvider, + ILogger logger) + : this(pskProvider, logger, DefaultGatedPrefixes) + { + } + + /// + /// Creates the interceptor gating an explicit prefix set. Internal — a public second + /// constructor would reintroduce the ambiguous-constructor defect described on the class. + /// + /// Resolves each site's preshared key. + /// Logger for denial diagnostics. + /// Method-path prefixes to gate. + internal CentralControlAuthInterceptor( + ISitePskProvider pskProvider, + ILogger logger, + IReadOnlyList gatedPrefixes) + { + ArgumentNullException.ThrowIfNull(pskProvider); + ArgumentNullException.ThrowIfNull(logger); + ArgumentNullException.ThrowIfNull(gatedPrefixes); + + _pskProvider = pskProvider; + _logger = logger; + _gatedPrefixes = gatedPrefixes; + } + + /// + public override async Task UnaryServerHandler( + TRequest request, + ServerCallContext context, + UnaryServerMethod continuation) + { + await AuthorizeAsync(context).ConfigureAwait(false); + return await continuation(request, context).ConfigureAwait(false); + } + + /// + public override async Task DuplexStreamingServerHandler( + IAsyncStreamReader requestStream, + IServerStreamWriter responseStream, + ServerCallContext context, + DuplexStreamingServerMethod continuation) + { + await AuthorizeAsync(context).ConfigureAwait(false); + await continuation(requestStream, responseStream, context).ConfigureAwait(false); + } + + /// + public override async Task ClientStreamingServerHandler( + IAsyncStreamReader requestStream, + ServerCallContext context, + ClientStreamingServerMethod continuation) + { + await AuthorizeAsync(context).ConfigureAwait(false); + return await continuation(requestStream, context).ConfigureAwait(false); + } + + /// + public override async Task ServerStreamingServerHandler( + TRequest request, + IServerStreamWriter responseStream, + ServerCallContext context, + ServerStreamingServerMethod continuation) + { + await AuthorizeAsync(context).ConfigureAwait(false); + await continuation(request, responseStream, context).ConfigureAwait(false); + } + + /// + /// Throws with unless a + /// gated call carries a valid x-scadabridge-site header AND a bearer token matching + /// that site's resolved key. Non-gated calls return immediately. + /// + private async Task AuthorizeAsync(ServerCallContext context) + { + if (!IsGated(context.Method)) + { + return; + } + + var siteId = ExtractSiteId(context.RequestHeaders); + if (string.IsNullOrWhiteSpace(siteId)) + { + _logger.LogWarning( + "Rejected a central control-plane call to {Method}: the required " + + "'{Header}' metadata header is missing or blank, so there is no per-site key to " + + "verify against.", + context.Method, ControlPlaneCredentials.SiteHeader); + throw Denied("missing site identity header"); + } + + string expected; + try + { + expected = await _pskProvider.GetAsync(siteId, context.CancellationToken) + .ConfigureAwait(false); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception ex) + { + // Fail-closed: an unresolvable key is a denial, never a pass-through. The provider + // has already logged the specific cause (missing secret / missing config entry). + _logger.LogWarning(ex, + "Rejected a central control-plane call to {Method}: no preshared key could be " + + "resolved for site {SiteId}.", + context.Method, siteId); + throw Denied("no key configured for the presented site"); + } + + var presented = ExtractBearerToken(context.RequestHeaders); + if (presented is null || !FixedTimeEquals(presented, expected)) + { + _logger.LogWarning( + "Rejected a central control-plane call to {Method} from site {SiteId}: {Reason}.", + context.Method, siteId, + presented is null ? "no bearer token presented" : "bearer token did not match"); + throw Denied("control plane authentication failed"); + } + } + + private static RpcException Denied(string reason) + => new(new Status(StatusCode.PermissionDenied, $"Control plane authentication failed: {reason}.")); + + private bool IsGated(string method) + { + foreach (var prefix in _gatedPrefixes) + { + if (method.StartsWith(prefix, StringComparison.Ordinal)) + { + return true; + } + } + return false; + } + + private static string? ExtractSiteId(Metadata headers) + { + foreach (var entry in headers) + { + if (string.Equals(entry.Key, ControlPlaneCredentials.SiteHeader, + StringComparison.OrdinalIgnoreCase)) + { + return entry.Value; + } + } + return null; + } + + private static string? ExtractBearerToken(Metadata headers) + { + // gRPC lowercases header keys on the wire; compare case-insensitively so a hand-built + // Metadata in a test behaves the same as a real request. + foreach (var entry in headers) + { + if (!string.Equals(entry.Key, ControlPlaneCredentials.AuthorizationHeader, + StringComparison.OrdinalIgnoreCase)) + { + continue; + } + + var value = entry.Value; + if (value is not null && value.StartsWith("Bearer ", StringComparison.OrdinalIgnoreCase)) + { + return value["Bearer ".Length..]; + } + } + + return null; + } + + private static bool FixedTimeEquals(string presented, string expected) + => CryptographicOperations.FixedTimeEquals( + Encoding.UTF8.GetBytes(presented), Encoding.UTF8.GetBytes(expected)); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/NodeOptions.cs b/src/ZB.MOM.WW.ScadaBridge.Host/NodeOptions.cs index a4dd0c39..3c004478 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/NodeOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/NodeOptions.cs @@ -21,6 +21,15 @@ public class NodeOptions /// Gets or sets the gRPC port for the site stream server. public int GrpcPort { get; set; } = 8083; /// + /// HTTP/2 (h2c) port the CENTRAL node listens on for the site→central + /// CentralControlService gRPC control plane. Default 8083 — deliberately symmetric + /// with the site , since a node is either central or a site and the + /// two never share a process. This listener is distinct from central's :5000 HTTP/1 + /// surface (Central UI, Management/Inbound API), which stays exactly as-is: gRPC does NOT go + /// through Traefik (HTTP/1 only). Ignored on site nodes. + /// + public int CentralGrpcPort { get; set; } = 8083; + /// /// HTTP/1.1 port serving the Prometheus /metrics scrape endpoint on site nodes. /// Defaults to 8084 — deliberately distinct from (8082) /// and (8083) so the Kestrel metrics listener never contends diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/NodeOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.Host/NodeOptionsValidator.cs index 1b5a7623..6890fa8a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/NodeOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/NodeOptionsValidator.cs @@ -23,6 +23,7 @@ public sealed class NodeOptionsValidator : OptionsValidatorBase RequirePort(builder, options.RemotingPort, nameof(NodeOptions.RemotingPort)); RequirePort(builder, options.GrpcPort, nameof(NodeOptions.GrpcPort)); RequirePort(builder, options.MetricsPort, nameof(NodeOptions.MetricsPort)); + RequirePort(builder, options.CentralGrpcPort, nameof(NodeOptions.CentralGrpcPort)); } // 0 stays valid (dynamic-port request); reject only out-of-TCP-range values. diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs index 9a576e15..2422f353 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs @@ -97,6 +97,24 @@ try // Windows Service support (no-op when not running as a Windows Service) builder.Host.UseWindowsService(); + // Explicit Kestrel h2c listener for the central-hosted gRPC control plane + // (CentralControlService). Mirrors the site branch's gRPC listener: HTTP/2-only, + // on its own port (default 8083, symmetric with the site GrpcPort — a node is + // either central or a site, never both). This is ADDITIVE to central's :5000 + // HTTP/1 surface (Central UI + Management/Inbound API from ASPNETCORE_URLS), + // which is untouched: gRPC does NOT go through Traefik (HTTP/1 only), sites reach + // this port by container name. T1A.2 hosts the server here; sites keep dialing over + // Akka ClusterClient until T1A.3 flips CentralTransport — this listener simply also + // accepts. + var centralGrpcPort = configuration.GetValue("ScadaBridge:Node:CentralGrpcPort", 8083); + builder.WebHost.ConfigureKestrel(options => + { + options.ListenAnyIP(centralGrpcPort, listenOptions => + { + listenOptions.Protocols = Microsoft.AspNetCore.Server.Kestrel.Core.HttpProtocols.Http2; + }); + }); + // Shared components builder.Services.AddClusterInfrastructure(); builder.Services.AddCommunication(); @@ -108,6 +126,22 @@ try // node uses for its single key). Registered before the clients that consume it. builder.Services.AddSingleton< ZB.MOM.WW.ScadaBridge.Communication.Grpc.ISitePskProvider, SitePskProvider>(); + + // Central-hosted gRPC control plane (T1A.2). CentralControlAuthInterceptor is the + // per-site sibling of the site's ControlPlaneAuthInterceptor: it verifies the Bearer + // token against the key for the site named in the required x-scadabridge-site header, + // resolved through the ISitePskProvider registered just above. Registered BY TYPE on + // AddGrpc exactly as the site branch registers its interceptors — NOT as a DI singleton, + // which would let DI hand the instance back and bypass Grpc.AspNetCore's own activation + // path (the shape that once hid a two-public-constructor defect until the rig caught it). + // CentralControlGrpcService decodes each request onto the same in-process message the + // ClusterClient path carries and Asks CentralCommunicationActor — zero handler logic here. + builder.Services.AddGrpc(options => + { + options.Interceptors.Add(); + }); + builder.Services.AddSingleton< + ZB.MOM.WW.ScadaBridge.Communication.Grpc.CentralControlGrpcService>(); builder.Services.AddHealthMonitoring(); builder.Services.AddCentralHealthAggregation(); builder.Services.AddExternalSystemGateway(); @@ -462,6 +496,14 @@ try // Requires endpoint routing (app.UseRouting() above). app.MapZbMetrics(); + // Central-hosted gRPC control plane (T1A.2) — the site→central CentralControlService. + // Runs on the dedicated h2c listener configured above (default :8083), gated by + // CentralControlAuthInterceptor and readiness-gated by the service itself (Unavailable + // until AkkaHostedService hands the CentralCommunicationActor over via SetReady). It + // shares the app's endpoint routing with the HTTP/1 surface; Kestrel steers each + // connection to the right pipeline by listener/protocol. + app.MapGrpcService(); + app.MapStaticAssets(); app.MapCentralUI(); app.MapInboundAPI(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlGrpcServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlGrpcServiceTests.cs new file mode 100644 index 00000000..9d2f0e2b --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlGrpcServiceTests.cs @@ -0,0 +1,181 @@ +using Akka.Actor; +using Akka.TestKit; +using Akka.TestKit.Xunit2; +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification; +using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc; + +/// +/// Behaviour of that does not need a real gRPC pipeline: +/// the readiness gate, the fire-and-forget heartbeat contract, the ingest-timeout status +/// mapping, and the shared audit-ingest timeout constant. The auth interceptor + real transport +/// are proven separately in CentralControlEndToEndTests (Host.Tests). +/// +public class CentralControlGrpcServiceTests : TestKit +{ + private static ServerCallContext NewContext(CancellationToken ct = default) + { + var context = Substitute.For(); + context.CancellationToken.Returns(ct); + return context; + } + + private CentralControlGrpcService CreateService(CommunicationOptions? options = null) + => new( + NullLogger.Instance, + Options.Create(options ?? new CommunicationOptions())); + + [Fact] + public async Task BeforeSetReady_AUnaryCall_IsUnavailable() + { + // Nothing was dispatched, so Unavailable is the right status — it tells a site transport + // the call never ran and a cross-node retry is safe. + var service = CreateService(); + + var ex = await Assert.ThrowsAsync( + () => service.SubmitNotification(new NotificationSubmitDto(), NewContext())); + + Assert.Equal(StatusCode.Unavailable, ex.StatusCode); + } + + [Fact] + public async Task BeforeSetReady_ANonEmptyIngestBatch_IsUnavailable() + { + var service = CreateService(); + var batch = new AuditEventBatch(); + batch.Events.Add(NewAuditDto()); + + var ex = await Assert.ThrowsAsync( + () => service.IngestAuditEvents(batch, NewContext())); + + Assert.Equal(StatusCode.Unavailable, ex.StatusCode); + } + + [Fact] + public async Task AfterSetReady_AUnaryCall_ReachesTheActor() + { + var stub = Sys.ActorOf(Props.Create(() => new StubActor())); + var service = CreateService(); + service.SetReady(stub); + + var ack = await service.SubmitNotification(NewNotificationDto(), NewContext()); + + Assert.True(ack.Accepted); + } + + [Fact] + public async Task Heartbeat_BeforeSetReady_SucceedsAndIsDropped() + { + // Fire-and-forget end-to-end: a heartbeat must never fault the site's timer, so even + // with no actor wired the call returns OK rather than Unavailable. + var service = CreateService(); + + var reply = await service.Heartbeat(NewHeartbeatDto(), NewContext()); + + Assert.NotNull(reply); + } + + [Fact] + public async Task Heartbeat_AfterSetReady_TellsTheActor_AndNeverAsks() + { + // A black-hole actor that never replies would hang an Ask forever; the heartbeat still + // returns immediately, proving it is a Tell, not an Ask. + var blackHole = Sys.ActorOf(Props.Create(() => new NeverRepliesActor())); + var service = CreateService(); + service.SetReady(blackHole); + + var reply = await service.Heartbeat(NewHeartbeatDto(), NewContext()); + + Assert.NotNull(reply); + } + + [Fact] + public async Task WhenTheActorNeverReplies_TheCall_IsDeadlineExceeded_NotUnavailable() + { + // The message WAS delivered (Unavailable would wrongly invite a duplicate retry on the + // peer node); a timeout is DeadlineExceeded, which callers never cross-node-retry. + var blackHole = Sys.ActorOf(Props.Create(() => new NeverRepliesActor())); + var service = CreateService(new CommunicationOptions + { + NotificationForwardTimeout = TimeSpan.FromMilliseconds(200), + }); + service.SetReady(blackHole); + + var ex = await Assert.ThrowsAsync( + () => service.SubmitNotification(NewNotificationDto(), NewContext())); + + Assert.Equal(StatusCode.DeadlineExceeded, ex.StatusCode); + } + + [Fact] + public async Task EmptyIngestBatch_ShortCircuits_EvenBeforeSetReady() + { + // An empty batch is a no-op the actor need never see; it must not depend on readiness. + var service = CreateService(); + + var ack = await service.IngestAuditEvents(new AuditEventBatch(), NewContext()); + + Assert.Empty(ack.AcceptedEventIds); + } + + [Fact] + public void TheAuditIngestTimeout_IsTheOneSharedConstant() + { + // The plan calls SiteStreamGrpcServer.AuditIngestAskTimeout "one source of truth" shared + // between the two audit-ingest transports; the central service must not re-declare 30s. + Assert.Equal(TimeSpan.FromSeconds(30), SiteStreamGrpcServer.AuditIngestAskTimeout); + } + + // The mapper reads SiteEnqueuedAt unconditionally, so a DTO that reaches mapping must carry + // a timestamp. Only DTOs that get past the readiness/auth gate map, so the negative tests + // above can pass a bare DTO. + private static NotificationSubmitDto NewNotificationDto() => new() + { + NotificationId = Guid.NewGuid().ToString(), + ListName = "ops", + SiteEnqueuedAt = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTime( + DateTime.SpecifyKind(new DateTime(2026, 5, 20, 10, 0, 0), DateTimeKind.Utc)), + }; + + private static HeartbeatDto NewHeartbeatDto() => new() + { + SiteId = "site-a", + NodeHostname = "node-a", + IsActive = true, + Timestamp = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTime( + DateTime.SpecifyKind(new DateTime(2026, 5, 20, 10, 0, 0), DateTimeKind.Utc)), + }; + + private static AuditEventDto NewAuditDto() => new() + { + EventId = Guid.NewGuid().ToString(), + OccurredAtUtc = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTime( + DateTime.SpecifyKind(new DateTime(2026, 5, 20, 10, 0, 0), DateTimeKind.Utc)), + Channel = "ApiOutbound", + Kind = "ApiCall", + Status = "Delivered", + SourceSiteId = "site-a", + }; + + private sealed class StubActor : ReceiveActor + { + public StubActor() + { + Receive(msg => + Sender.Tell(new NotificationSubmitAck(msg.NotificationId, Accepted: true, Error: null))); + } + } + + /// Swallows every message and never replies, so an Ask against it times out. + private sealed class NeverRepliesActor : ReceiveActor + { + public NeverRepliesActor() => ReceiveAny(_ => { }); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralControlAuthInterceptorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralControlAuthInterceptorTests.cs new file mode 100644 index 00000000..9e642ab7 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralControlAuthInterceptorTests.cs @@ -0,0 +1,197 @@ +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Host.Tests; + +/// +/// Central's inbound gate for the site→central gRPC control plane (T1A.2). The mirror of +/// , but central's verification model is inverted: +/// it looks up the key for the site named in the x-scadabridge-site header and checks the +/// bearer token against THAT, rather than against one own-key. +/// +/// +/// Central genuinely needs a distinct verification model, so it is a separate class with its own +/// single public constructor — not a variant ctor on , +/// whose two-public-constructor form once silently disabled the gate. The one-public-ctor +/// invariant is pinned below exactly as the sibling pins it. +/// +public class CentralControlAuthInterceptorTests +{ + private const string ControlMethod = + "/scadabridge.centralcontrol.v1.CentralControlService/SubmitNotification"; + private const string SiteStreamMethod = "/sitestream.SiteStreamService/SubscribeInstance"; + + private const string SiteA = "site-a"; + private const string SiteAKey = "site-a-preshared-key"; + private const string SiteB = "site-b"; + private const string SiteBKey = "site-b-preshared-key"; + + /// An backed by a fixed site→key map; throws for unknown sites. + private sealed class MapPskProvider(IReadOnlyDictionary keys) : ISitePskProvider + { + public ValueTask GetAsync(string siteId, CancellationToken ct) + => keys.TryGetValue(siteId, out var key) + ? new ValueTask(key) + : throw new InvalidOperationException($"no key for '{siteId}'"); + + public void Invalidate(string siteId) { } + } + + private static CentralControlAuthInterceptor CreateInterceptor() + => new( + new MapPskProvider(new Dictionary { [SiteA] = SiteAKey, [SiteB] = SiteBKey }), + NullLogger.Instance); + + private static ServerCallContext CreateContext( + string method, string? siteHeader, string? authorizationHeader) + { + var headers = new Metadata(); + if (siteHeader is not null) + headers.Add(ControlPlaneCredentials.SiteHeader, siteHeader); + if (authorizationHeader is not null) + headers.Add(ControlPlaneCredentials.AuthorizationHeader, authorizationHeader); + + return new FakeServerCallContext(method, headers); + } + + private sealed class FakeServerCallContext(string method, Metadata requestHeaders) + : ServerCallContext + { + protected override string MethodCore => method; + protected override string HostCore => "localhost"; + protected override string PeerCore => "ipv4:127.0.0.1:12345"; + protected override DateTime DeadlineCore => DateTime.UtcNow.AddMinutes(1); + protected override Metadata RequestHeadersCore => requestHeaders; + protected override CancellationToken CancellationTokenCore => CancellationToken.None; + protected override Metadata ResponseTrailersCore { get; } = []; + protected override Status StatusCore { get; set; } + protected override WriteOptions? WriteOptionsCore { get; set; } + protected override AuthContext AuthContextCore { get; } = + new(null, new Dictionary>()); + + protected override ContextPropagationToken CreatePropagationTokenCore( + ContextPropagationOptions? options) + => throw new NotSupportedException(); + + protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) + => Task.CompletedTask; + } + + private static Task Invoke( + CentralControlAuthInterceptor interceptor, ServerCallContext context) + => interceptor.UnaryServerHandler( + "request", context, (_, _) => Task.FromResult("ok")); + + [Fact] + public async Task NonGatedMethod_PassesThrough() + { + // Central's interceptor gates only CentralControlService; anything else on the listener + // (there is nothing today) is not its concern. + var interceptor = CreateInterceptor(); + var context = CreateContext(SiteStreamMethod, siteHeader: null, authorizationHeader: null); + + Assert.Equal("ok", await Invoke(interceptor, context)); + } + + [Fact] + public async Task CorrectSiteAndKey_IsAccepted() + { + var interceptor = CreateInterceptor(); + var context = CreateContext(ControlMethod, SiteA, $"Bearer {SiteAKey}"); + + Assert.Equal("ok", await Invoke(interceptor, context)); + } + + [Fact] + public async Task MissingSiteHeader_IsDenied() + { + // Fail-closed: no header means no per-site key to verify against, so there is nothing to + // pass through TO. A present bearer token does not rescue it. + var interceptor = CreateInterceptor(); + var context = CreateContext(ControlMethod, siteHeader: null, $"Bearer {SiteAKey}"); + + var ex = await Assert.ThrowsAsync(() => Invoke(interceptor, context)); + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task BlankSiteHeader_IsDenied() + { + var interceptor = CreateInterceptor(); + var context = CreateContext(ControlMethod, siteHeader: " ", $"Bearer {SiteAKey}"); + + var ex = await Assert.ThrowsAsync(() => Invoke(interceptor, context)); + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task UnknownSite_IsDenied_NotPassedThrough() + { + // The provider throws for an unknown site; the interceptor must turn that into a denial, + // never swallow it and let the call proceed unauthenticated. + var interceptor = CreateInterceptor(); + var context = CreateContext(ControlMethod, "site-nonexistent", "Bearer whatever"); + + var ex = await Assert.ThrowsAsync(() => Invoke(interceptor, context)); + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task NoBearerToken_IsDenied() + { + var interceptor = CreateInterceptor(); + var context = CreateContext(ControlMethod, SiteA, authorizationHeader: null); + + var ex = await Assert.ThrowsAsync(() => Invoke(interceptor, context)); + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task WrongKeyForTheSite_IsDenied() + { + // site-a presents site-b's key. Both keys are valid keys; the point is the token must + // match the key for THIS site, not just be a key central knows. + var interceptor = CreateInterceptor(); + var context = CreateContext(ControlMethod, SiteA, $"Bearer {SiteBKey}"); + + var ex = await Assert.ThrowsAsync(() => Invoke(interceptor, context)); + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task EachSiteIsVerifiedAgainstItsOwnKey() + { + // site-b with site-b's key is accepted by the same interceptor instance that rejected + // site-a-with-site-b's-key above — proving the per-site lookup, not a single shared key. + var interceptor = CreateInterceptor(); + var context = CreateContext(ControlMethod, SiteB, $"Bearer {SiteBKey}"); + + Assert.Equal("ok", await Invoke(interceptor, context)); + } + + [Fact] + public void TheInterceptorHasExactlyOnePublicConstructor() + { + // Grpc.AspNetCore registers this interceptor BY TYPE; InterceptorRegistration.GetFactory() + // throws "Multiple constructors accepting all given argument types have been found" the + // moment a second public constructor is applicable, and the throw lands inside the + // pipeline on every call — a gate that authorizes nothing while looking like a handler + // bug. The explicit-prefix constructor is internal to keep it from recurring; this pins + // that. (Same invariant as ControlPlaneAuthInterceptorTests.) + var publicCtors = typeof(CentralControlAuthInterceptor).GetConstructors(); + + Assert.Single(publicCtors); + } + + [Fact] + public void DefaultGatedPrefixes_MatchTheRealCentralControlServicePath() + { + // A typo here disables the whole gate silently: every call would pass through. Pin it + // against the generated service descriptor, not the proto text. + var method = CentralControlService.Descriptor.FullName; + Assert.Contains( + CentralControlAuthInterceptor.DefaultGatedPrefixes, + p => p == $"/{method}/"); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralControlEndToEndTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralControlEndToEndTests.cs new file mode 100644 index 00000000..63d806aa --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralControlEndToEndTests.cs @@ -0,0 +1,256 @@ +using Akka.Actor; +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification; +using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Host.Tests; + +/// +/// End-to-end proof of the central-hosted gRPC control plane (T1A.2): the real +/// AND the real +/// over a real gRPC stack, the service Asking a real +/// (stub) CentralCommunicationActor stand-in. +/// +/// +/// +/// The interceptor is registered BY TYPE on AddGrpc, exactly as Program.cs +/// does, and is deliberately NOT placed in DI as a singleton — pre-registering it lets DI hand +/// the instance back and bypasses Grpc.AspNetCore's own activation path, which is how the +/// two-public-constructor defect escaped a green suite once before. This harness copies the +/// shape of for the same reason. +/// +/// +/// Runs in-process over : no ports, no containers. A tiny Akka actor +/// stands in for CentralCommunicationActor so the test never touches MSSQL or a cluster; +/// the method paths, message types and mapper are the real generated/production ones. +/// +/// +public class CentralControlEndToEndTests : IAsyncLifetime +{ + private const string SiteA = "site-a"; + private const string SiteAKey = "site-a-preshared-key"; + private const string SiteB = "site-b"; + private const string SiteBKey = "site-b-preshared-key"; + + // The deterministic id the stub actor "accepts" for every ingest batch, so the ingest + // bridge can be asserted without extracting ids out of a decoded AuditEvent. + private static readonly Guid AcceptedId = Guid.Parse("11111111-1111-1111-1111-111111111111"); + + private IHost _host = null!; + private TestServer _server = null!; + private ActorSystem _actorSystem = null!; + private CentralControlGrpcService _service = null!; + + /// + public async Task InitializeAsync() + { + _actorSystem = ActorSystem.Create("centralcontrol-e2e-test"); + var stub = _actorSystem.ActorOf(Props.Create(() => new StubCentralActor(AcceptedId)), "central-stub"); + + _service = new CentralControlGrpcService( + NullLogger.Instance, + Options.Create(new CommunicationOptions())); + _service.SetReady(stub); + + var pskProvider = new MapPskProvider(new Dictionary + { + [SiteA] = SiteAKey, + [SiteB] = SiteBKey, + }); + + _host = await new HostBuilder() + .ConfigureWebHost(web => web + .UseTestServer() + .ConfigureServices(services => + { + // BY TYPE, and NOT also in DI — see the class remarks. + services.AddGrpc(o => o.Interceptors.Add()); + services.AddSingleton(pskProvider); + services.AddSingleton(_service); + }) + .Configure(app => + { + app.UseRouting(); + app.UseEndpoints(e => e.MapGrpcService()); + })) + .StartAsync(); + + _server = _host.GetTestServer(); + } + + /// + public async Task DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + await _actorSystem.Terminate(); + } + + /// Builds a channel credentialed for with . + private GrpcChannel Channel(string? key, string siteId) + { + var options = new GrpcChannelOptions { HttpHandler = _server.CreateHandler() }; + if (key is not null) + { + options.WithSiteCredentials(new FixedPskProvider(key), siteId); + } + return GrpcChannel.ForAddress(_server.BaseAddress, options); + } + + private CentralControlService.CentralControlServiceClient Client(string? key, string siteId) + => new(Channel(key, siteId)); + + // ---- Auth positives / negatives, all through the real pipeline ---- + + [Fact] + public async Task CorrectSiteAndKey_ReachesTheService_OnAUnaryCall() + { + var client = Client(SiteAKey, SiteA); + + var ack = await client.SubmitNotificationAsync(NewNotificationDto()); + + Assert.True(ack.Accepted); + } + + [Fact] + public async Task NoCredentialsAtAll_IsRejected_WithPermissionDenied() + { + var client = Client(key: null, SiteA); + + var ex = await Assert.ThrowsAsync( + async () => await client.SubmitNotificationAsync(new NotificationSubmitDto())); + + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task WrongKeyForTheSite_IsRejected_WithPermissionDenied() + { + // site-a presents site-b's (valid, but wrong-for-this-site) key. + var client = Client(SiteBKey, SiteA); + + var ex = await Assert.ThrowsAsync( + async () => await client.SubmitNotificationAsync(new NotificationSubmitDto())); + + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task UnknownSite_IsRejected_WithPermissionDenied() + { + var client = Client("any-key", "site-nonexistent"); + + var ex = await Assert.ThrowsAsync( + async () => await client.SubmitNotificationAsync(new NotificationSubmitDto())); + + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task TheAcceptAndRejectPathsAreDistinguishable() + { + // A gate whose accept and reject paths produce the same observable result is not a gate. + // Correct key → the service answers (Accepted:true); wrong key → PermissionDenied, + // never reaching the service. These are two different outcomes, which is the whole point. + var ok = await Client(SiteAKey, SiteA).SubmitNotificationAsync(NewNotificationDto()); + Assert.True(ok.Accepted); + + var ex = await Assert.ThrowsAsync( + async () => await Client("wrong", SiteA).SubmitNotificationAsync(new NotificationSubmitDto())); + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + // ---- One RPC per shape: unary (above) + the ingest bridge ---- + + [Fact] + public async Task IngestAuditEvents_DecodesTheBatch_AsksTheActor_EncodesTheAck() + { + var client = Client(SiteAKey, SiteA); + + var batch = new AuditEventBatch(); + batch.Events.Add(NewAuditDto()); + batch.Events.Add(NewAuditDto()); + + var ack = await client.IngestAuditEventsAsync(batch); + + // The stub actor accepts one deterministic id per non-empty batch; its presence proves + // the full DTO→command→Ask→reply→ack bridge ran, gated call and all. + Assert.Contains(AcceptedId.ToString(), ack.AcceptedEventIds); + } + + [Fact] + public async Task IngestAuditEvents_EmptyBatch_ShortCircuits_WithoutAskingTheActor() + { + // Even the empty-batch fast path is behind the gate — it still needs a valid key. + var client = Client(SiteAKey, SiteA); + + var ack = await client.IngestAuditEventsAsync(new AuditEventBatch()); + + Assert.Empty(ack.AcceptedEventIds); + } + + // The mapper reads SiteEnqueuedAt unconditionally; only DTOs that clear the gate reach it, + // so the accept-path tests carry a timestamp while the negative tests can pass a bare DTO. + private static NotificationSubmitDto NewNotificationDto() => new() + { + NotificationId = Guid.NewGuid().ToString(), + ListName = "ops", + SiteEnqueuedAt = Timestamp.FromDateTime( + DateTime.SpecifyKind(new DateTime(2026, 5, 20, 10, 0, 0), DateTimeKind.Utc)), + }; + + private static AuditEventDto NewAuditDto() => new() + { + EventId = Guid.NewGuid().ToString(), + OccurredAtUtc = Timestamp.FromDateTime( + DateTime.SpecifyKind(new DateTime(2026, 5, 20, 10, 0, 0), DateTimeKind.Utc)), + Channel = "ApiOutbound", + Kind = "ApiCall", + Status = "Delivered", + SourceSiteId = SiteA, + }; + + private sealed class FixedPskProvider(string key) : ISitePskProvider + { + public ValueTask GetAsync(string siteId, CancellationToken ct) => new(key); + public void Invalidate(string siteId) { } + } + + private sealed class MapPskProvider(IReadOnlyDictionary keys) : ISitePskProvider + { + public ValueTask GetAsync(string siteId, CancellationToken ct) + => keys.TryGetValue(siteId, out var key) + ? new ValueTask(key) + : throw new InvalidOperationException($"no key for '{siteId}'"); + + public void Invalidate(string siteId) { } + } + + /// + /// Minimal stand-in for CentralCommunicationActor: answers the two RPC shapes this + /// test exercises. Replies straight to the Ask's temp sender, exactly as the real actor's + /// Forward/PipeTo paths do. + /// + private sealed class StubCentralActor : ReceiveActor + { + public StubCentralActor(Guid acceptedId) + { + Receive(msg => + Sender.Tell(new NotificationSubmitAck(msg.NotificationId, Accepted: true, Error: null))); + + Receive(_ => + Sender.Tell(new IngestAuditEventsReply(new[] { acceptedId }))); + } + } +}