From 59c695191cd9d980daf302d6932f9051273f624a Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 19 Jul 2026 09:29:22 -0400 Subject: [PATCH] feat(localdb): fail-closed auth on the sync endpoint + replication health signal Tasks 8 and 9 of the LocalDb Phase 1 adoption plan. Task 8 - LocalDbSyncAuthInterceptor. The replication library's LocalDbSyncService verifies nothing; inbound auth is explicitly the host's job. Without this, anything able to reach a site node's gRPC port could stream arbitrary rows into the consolidated site database - including OperationTracking, which central reconciles from. Scoped strictly to /localdb_sync.v1.LocalDbSync/; SiteStream shares the same AddGrpc pipeline and passes through untouched. Fail-closed: with no LocalDb:Replication:ApiKey configured NO sync stream is accepted, authenticated or not. That is deliberate - "no key" is the default every site node ships with, so treating it as "no auth required" would expose the endpoint on precisely the most common configuration. Comparison is FixedTimeEquals over UTF-8 bytes. All four server handler shapes are gated, not just unary: the sync RPC is a bidirectional stream, so gating only the unary path would leave the real endpoint open while every unary test still passed. There is a test for that. Deviation from the plan: it specified Grpc.Core.Testing for the fake ServerCallContext. That type ships in the retired native Grpc.Core package and does not exist on the grpc-dotnet stack this solution uses; a minimal FakeServerCallContext in the test file was the better trade than adding a dead dependency. Task 9 - ISyncStatus onto the site health report as LocalDbReplicationConnected and LocalDbOplogBacklog, via a delegate-seam hosted service following the AddSiteEventLogHealthMetricsBridge precedent (HealthMonitoring takes no reference on the replication library). Both are additive init properties, so the Akka-remoted SiteHealthReport constructor signature is untouched. Both fields are nullable and the distinction is load-bearing: - null = the reporter has not run / replication is not wired ("no data"); - false/0 = a real reading. On a node with no peer that IS the healthy default-OFF state, not an outage. OplogBacklog is passed through nullable end-to-end because ISyncStatus returns null when the poll fails - flattening it to 0 would report a replication pair that cannot read its own oplog as perfectly healthy. The collector stores both values as one tuple and CollectReport reads it once, so a torn read cannot pair a fresh Connected with a stale backlog. Verified: build 0 warnings; Host 307/307 (8 interceptor + 3 health tests new), HealthMonitoring 97/97, Commons 684/684. Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts --- .../Messages/Health/SiteHealthReport.cs | 31 ++++ .../ISiteHealthCollector.cs | 19 ++ .../LocalDbReplicationStatusReporter.cs | 130 +++++++++++++ .../ServiceCollectionExtensions.cs | 46 +++++ .../SiteHealthCollector.cs | 20 +- .../LocalDbSyncAuthInterceptor.cs | 158 ++++++++++++++++ src/ZB.MOM.WW.ScadaBridge.Host/Program.cs | 11 +- .../SiteServiceRegistration.cs | 10 + .../LocalDbSyncAuthInterceptorTests.cs | 171 ++++++++++++++++++ .../SiteLocalDbWiringTests.cs | 46 +++++ 10 files changed, 637 insertions(+), 5 deletions(-) create mode 100644 src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/LocalDbReplicationStatusReporter.cs create mode 100644 src/ZB.MOM.WW.ScadaBridge.Host/LocalDbSyncAuthInterceptor.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Host.Tests/LocalDbSyncAuthInterceptorTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/SiteHealthReport.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/SiteHealthReport.cs index 216bd656..5e498b78 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/SiteHealthReport.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/SiteHealthReport.cs @@ -73,6 +73,37 @@ public record SiteHealthReport( /// indicates a stuck/blocked script holding a dedicated thread. /// public double? ScriptOldestBusyAgeSeconds { get; init; } + + // LocalDb 2-node replication of the consolidated site database (Phase 1). + // Additive init properties for the same reason as the scheduler gauges above: + // the positional constructor stays untouched. Refreshed on the site by + // LocalDbReplicationStatusReporter. + + /// + /// Whether a replication sync session is currently running with the peer site node, + /// or when replication is not wired on this node. + /// + /// + /// Nullable on purpose, and is NOT the same as null. + /// Replication ships default-OFF, so a site node with no peer configured reports + /// — that is a healthy, expected state, not an outage. Only + /// a node that was configured with a peer and is reporting + /// is degraded, and distinguishing those two cases is the operator's job, not this + /// field's. + /// + public bool? LocalDbReplicationConnected { get; init; } + + /// + /// Unacked replication oplog entries waiting to reach the peer, or + /// when unknown. + /// + /// + /// Null means unknown, never zero. The underlying provider returns null when + /// no backlog source is wired or the poll failed, and that must survive to the + /// operator: a failed read rendered as "0 backlog" would report a broken replication + /// pair as perfectly healthy. Collapsing null to 0 anywhere on this path is a bug. + /// + public long? LocalDbOplogBacklog { get; init; } } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ISiteHealthCollector.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ISiteHealthCollector.cs index 3f9ab84d..8781553a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ISiteHealthCollector.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ISiteHealthCollector.cs @@ -141,6 +141,25 @@ public interface ISiteHealthCollector // SiteHealthCollector overrides this with the Interlocked.Exchange store. } + /// + /// Replace the latest LocalDb replication status (peer-session connectivity and + /// unacked oplog backlog) used by the next call. + /// Refreshed periodically by the LocalDbReplicationStatusReporter hosted + /// service. Point-in-time: values are NOT reset on . + /// + /// + /// Whether a sync session is currently running. is the + /// normal state on a node with no peer configured — replication ships default-OFF. + /// + /// + /// Unacked oplog entries, or when unknown. Pass null through + /// unchanged: a failed poll rendered as 0 would report a broken pair as healthy. + /// + void SetLocalDbReplicationStatus(bool connected, long? oplogBacklog) + { + // Default no-op so test fakes do not need to be updated. + } + /// /// Replace the latest script-execution-scheduler gauges (queue depth, busy /// thread count, and age in seconds of the oldest in-flight script) used by diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/LocalDbReplicationStatusReporter.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/LocalDbReplicationStatusReporter.cs new file mode 100644 index 00000000..39f14ff1 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/LocalDbReplicationStatusReporter.cs @@ -0,0 +1,130 @@ +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring; + +/// +/// Site-side hosted service that periodically reads LocalDb replication status and pushes +/// it into , so the next +/// emits fresh +/// LocalDbReplicationConnected / LocalDbOplogBacklog fields. +/// +/// +/// +/// Why delegates and not ISyncStatus directly. Same reasoning as +/// : HealthMonitoring does not take a +/// reference on the replication library. The Host site wiring captures the two reads as +/// lambdas at registration time; this service only moves numbers. +/// +/// +/// Null backlog is preserved, never coerced to zero. ISyncStatus.OplogBacklog +/// is nullable precisely so a failed poll reads as "unknown" — rendering it as 0 would +/// report a replication pair that cannot read its own oplog as perfectly healthy. +/// +/// +/// Cadence. 30 s, matching and +/// SiteAuditBacklogReporter. Any exception during a probe is logged and swallowed; +/// the next tick retries. +/// +/// +public sealed class LocalDbReplicationStatusReporter : IHostedService, IDisposable +{ + /// Default poll cadence, matching the other site health bridges. + internal static readonly TimeSpan DefaultRefreshInterval = TimeSpan.FromSeconds(30); + + private readonly Func _connectedProvider; + private readonly Func _oplogBacklogProvider; + private readonly ISiteHealthCollector _collector; + private readonly ILogger _logger; + private readonly TimeSpan _refreshInterval; + private CancellationTokenSource? _cts; + private Task? _loop; + + /// Initializes a new instance of . + /// Reads whether a peer sync session is currently running. + /// Reads the unacked oplog backlog, or null when unknown. + /// The site health collector receiving the snapshot. + /// Logger instance. + /// Poll interval override; defaults to 30 s. + public LocalDbReplicationStatusReporter( + Func connectedProvider, + Func oplogBacklogProvider, + ISiteHealthCollector collector, + ILogger logger, + TimeSpan? refreshInterval = null) + { + _connectedProvider = connectedProvider ?? throw new ArgumentNullException(nameof(connectedProvider)); + _oplogBacklogProvider = oplogBacklogProvider ?? throw new ArgumentNullException(nameof(oplogBacklogProvider)); + _collector = collector ?? throw new ArgumentNullException(nameof(collector)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + _refreshInterval = refreshInterval ?? DefaultRefreshInterval; + } + + /// Starts the polling loop, probing once immediately before entering the timed cycle. + /// Cancellation token signalling host shutdown. + /// A task that represents the asynchronous operation. + public Task StartAsync(CancellationToken ct) + { + var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); + _cts = cts; + _loop = RunAsync(cts.Token); + return Task.CompletedTask; + } + + /// Stops the polling loop. + /// Cancellation token for the stop operation. + /// A task that represents the asynchronous operation. + public async Task StopAsync(CancellationToken ct) + { + if (_cts is null) return; + + await _cts.CancelAsync().ConfigureAwait(false); + if (_loop is not null) + { + // Await the loop so the reporter is quiescent before the host disposes the + // collector out from under it. + try { await _loop.ConfigureAwait(false); } + catch (OperationCanceledException) { /* expected on shutdown */ } + } + } + + private async Task RunAsync(CancellationToken ct) + { + while (!ct.IsCancellationRequested) + { + Probe(); + + try + { + await Task.Delay(_refreshInterval, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + return; + } + } + } + + /// + /// Reads both providers and pushes one snapshot onto the collector; never throws. + /// Public so a test can drive a single deterministic probe instead of starting the + /// service and waiting out the 30 s cadence. + /// + public void Probe() + { + try + { + _collector.SetLocalDbReplicationStatus( + _connectedProvider(), _oplogBacklogProvider()); + } + catch (Exception ex) + { + // Health reporting must never take the site node down. The previous snapshot + // stays on the collector and the next tick retries. + _logger.LogWarning(ex, "Failed to read LocalDb replication status for the site health report."); + } + } + + /// + public void Dispose() => _cts?.Dispose(); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ServiceCollectionExtensions.cs index e4587f24..d8d3b457 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ServiceCollectionExtensions.cs @@ -21,6 +21,12 @@ public static class ServiceCollectionExtensions /// private sealed class SiteEventLogHealthMetricsBridgeMarker { } + /// + /// Sentinel marker for 's idempotency + /// guard — same rationale as . + /// + private sealed class LocalDbReplicationHealthBridgeMarker { } + /// /// Register site-side health monitoring services (metric collection + periodic reporting). /// Call this on site nodes only. For central, call AddCentralHealthAggregation() instead. @@ -153,6 +159,46 @@ public static class ServiceCollectionExtensions return services; } + /// + /// Bridge LocalDb replication status (peer connectivity + unacked oplog backlog) onto + /// the site health report. Must be called AFTER AddSiteHealthMonitoring + /// (registers ) and after the replication engine is + /// registered. Idempotent via a marker sentinel. + /// + /// + /// Delegates rather than an ISyncStatus parameter keep HealthMonitoring free of + /// a reference on the replication library, matching + /// . Pass the backlog through as + /// nullable — coercing null to 0 would report an unreadable oplog as a healthy one. + /// + /// The service collection to register into. + /// Given the root provider, returns a reader for "a sync session is running". + /// Given the root provider, returns a reader for the unacked backlog (null = unknown). + /// The same for chaining. + public static IServiceCollection AddLocalDbReplicationHealthBridge( + this IServiceCollection services, + Func> connectedProvider, + Func> oplogBacklogProvider) + { + ArgumentNullException.ThrowIfNull(services); + ArgumentNullException.ThrowIfNull(connectedProvider); + ArgumentNullException.ThrowIfNull(oplogBacklogProvider); + + if (services.Any(d => d.ServiceType == typeof(LocalDbReplicationHealthBridgeMarker))) + { + return services; + } + + services.AddSingleton(); + services.AddHostedService(sp => new LocalDbReplicationStatusReporter( + connectedProvider(sp), + oplogBacklogProvider(sp), + sp.GetRequiredService(), + sp.GetRequiredService>())); + + return services; + } + /// /// Register the /// so a misconfigured ScadaBridge:HealthMonitoring section (zero/negative diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthCollector.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthCollector.cs index 51f22a7e..be008182 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthCollector.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthCollector.cs @@ -18,6 +18,10 @@ public class SiteHealthCollector : ISiteHealthCollector private int _auditRedactionFailures; private volatile SiteAuditBacklogSnapshot? _siteAuditBacklog; private long _siteEventLogWriteFailures; + // One volatile tuple rather than two independent fields: a torn read that paired a + // fresh Connected with a stale backlog would be indistinguishable from a real state. + // Null = the reporter has not yet run (or replication is not wired). + private volatile Tuple? _localDbReplicationStatus; private readonly ConcurrentDictionary _connectionStatuses = new(); private readonly ConcurrentDictionary _tagResolutionCounts = new(); private readonly ConcurrentDictionary _connectionEndpoints = new(); @@ -94,6 +98,12 @@ public class SiteHealthCollector : ISiteHealthCollector Interlocked.Exchange(ref _siteEventLogWriteFailures, count); } + /// + public void SetLocalDbReplicationStatus(bool connected, long? oplogBacklog) + { + _localDbReplicationStatus = Tuple.Create(connected, oplogBacklog); + } + /// public void UpdateConnectionHealth(string connectionName, ConnectionHealth health) { @@ -219,6 +229,9 @@ public class SiteHealthCollector : ISiteHealthCollector var deadLetters = Interlocked.Exchange(ref _deadLetterCount, 0); var siteAuditWriteFailures = Interlocked.Exchange(ref _siteAuditWriteFailures, 0); var auditRedactionFailures = Interlocked.Exchange(ref _auditRedactionFailures, 0); + // Single read of the volatile tuple — two reads could straddle a reporter tick and + // pair a fresh Connected with a stale backlog. + var localDbReplication = _localDbReplicationStatus; // Snapshot current connection and tag resolution state var connectionStatuses = new Dictionary(_connectionStatuses); @@ -259,7 +272,12 @@ public class SiteHealthCollector : ISiteHealthCollector { ScriptQueueDepth = Interlocked.CompareExchange(ref _scriptQueueDepth, 0, 0), ScriptBusyThreads = Interlocked.CompareExchange(ref _scriptBusyThreads, 0, 0), - ScriptOldestBusyAgeSeconds = ReadScriptOldestBusyAgeSeconds() + ScriptOldestBusyAgeSeconds = ReadScriptOldestBusyAgeSeconds(), + // Both fields come from the ONE snapshot read above. Null (the reporter has + // not run) leaves both report fields null — "no data", not "disconnected + // with an empty backlog". + LocalDbReplicationConnected = localDbReplication?.Item1, + LocalDbOplogBacklog = localDbReplication?.Item2 }; } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/LocalDbSyncAuthInterceptor.cs b/src/ZB.MOM.WW.ScadaBridge.Host/LocalDbSyncAuthInterceptor.cs new file mode 100644 index 00000000..d7b21d07 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Host/LocalDbSyncAuthInterceptor.cs @@ -0,0 +1,158 @@ +using System.Security.Cryptography; +using System.Text; +using Grpc.Core; +using Grpc.Core.Interceptors; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.LocalDb.Replication; + +namespace ZB.MOM.WW.ScadaBridge.Host; + +/// +/// Gates the LocalDb passive sync endpoint. The replication library deliberately leaves +/// inbound authentication to the host — its LocalDbSyncService verifies nothing — +/// so without this interceptor anything that can reach the site node's gRPC port could +/// stream arbitrary rows into the consolidated site database. +/// +/// +/// +/// Scoped by method path. Only calls under +/// /localdb_sync.v1.LocalDbSync/ are gated; every other method — notably the +/// existing SiteStream service sharing this listener — passes through untouched. +/// +/// +/// Fail-closed. With no LocalDb:Replication:ApiKey configured, NO sync +/// stream is accepted, authenticated or not. That is the deliberate choice: the +/// alternative — treating "no key" as "no auth required" — would silently expose the +/// endpoint on exactly the default configuration every site node ships with. An operator +/// enabling replication must set the same key on both nodes, which is already required +/// for the initiator to dial out (SyncBackgroundService sends +/// Authorization: Bearer <key>). +/// +/// +/// Comparison is over UTF-8 bytes, +/// so a wrong key cannot be recovered byte-by-byte from response timing. Length differences +/// are unavoidably observable and are not sensitive. +/// +/// +public sealed class LocalDbSyncAuthInterceptor : Interceptor +{ + private const string ServicePrefix = "/localdb_sync.v1.LocalDbSync/"; + private const string AuthorizationHeader = "authorization"; + private const string BearerPrefix = "Bearer "; + + private readonly IOptions _options; + private readonly ILogger _logger; + + /// Creates the interceptor. + /// Replication options; ApiKey is the expected bearer token. + /// Logger for denial diagnostics. + public LocalDbSyncAuthInterceptor( + IOptions options, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(logger); + + _options = options; + _logger = logger; + } + + /// + public override Task UnaryServerHandler( + TRequest request, + ServerCallContext context, + UnaryServerMethod continuation) + { + Authorize(context); + return continuation(request, context); + } + + /// + public override Task DuplexStreamingServerHandler( + IAsyncStreamReader requestStream, + IServerStreamWriter responseStream, + ServerCallContext context, + DuplexStreamingServerMethod continuation) + { + Authorize(context); + return continuation(requestStream, responseStream, context); + } + + /// + public override Task ClientStreamingServerHandler( + IAsyncStreamReader requestStream, + ServerCallContext context, + ClientStreamingServerMethod continuation) + { + Authorize(context); + return continuation(requestStream, context); + } + + /// + public override Task ServerStreamingServerHandler( + TRequest request, + IServerStreamWriter responseStream, + ServerCallContext context, + ServerStreamingServerMethod continuation) + { + Authorize(context); + return continuation(request, responseStream, context); + } + + /// + /// Throws with if + /// this is a sync call that does not carry the configured bearer token. Non-sync calls + /// return immediately. + /// + private void Authorize(ServerCallContext context) + { + if (!context.Method.StartsWith(ServicePrefix, StringComparison.Ordinal)) + return; + + var expected = _options.Value.ApiKey; + if (string.IsNullOrEmpty(expected)) + { + _logger.LogWarning( + "Rejected a LocalDb sync call to {Method}: no LocalDb:Replication:ApiKey is configured, " + + "so the passive sync endpoint is closed. Configure the same key on both nodes of the pair.", + context.Method); + throw new RpcException(new Status( + StatusCode.PermissionDenied, + "LocalDb sync is not accepting connections: no API key is configured on this node.")); + } + + var presented = ExtractBearerToken(context.RequestHeaders); + if (presented is null || !FixedTimeEquals(presented, expected)) + { + _logger.LogWarning( + "Rejected a LocalDb sync call to {Method}: {Reason}.", + context.Method, + presented is null ? "no bearer token presented" : "bearer token did not match"); + throw new RpcException(new Status( + StatusCode.PermissionDenied, + "LocalDb sync authentication failed.")); + } + } + + private static string? ExtractBearerToken(Metadata headers) + { + // gRPC lowercases header keys on the wire; compare case-insensitively anyway 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, AuthorizationHeader, StringComparison.OrdinalIgnoreCase)) + continue; + + var value = entry.Value; + if (value is not null && value.StartsWith(BearerPrefix, StringComparison.OrdinalIgnoreCase)) + return value[BearerPrefix.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/Program.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs index e8ea49ab..5df284bc 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs @@ -514,7 +514,11 @@ try }); // gRPC server registration - builder.Services.AddGrpc(); + // The interceptor gates ONLY /localdb_sync.v1.LocalDbSync/ — SiteStream calls on + // this same pipeline pass through untouched. It is fail-closed: with no + // LocalDb:Replication:ApiKey configured, no sync stream is accepted at all. + builder.Services.AddGrpc(options => + options.Interceptors.Add()); builder.Services.AddSingleton(); // Existing site service registrations (this is also where LocalDb and its @@ -540,9 +544,8 @@ try // The passive half of LocalDb replication: the peer node dials THIS endpoint. // It shares the Kestrel h2c listener the site gRPC server already uses, so no // listener or port changes are needed. Mapping it is harmless with no peer - // configured — nothing dials it. Inbound auth is the host's job and lands in - // Task 8; until then this endpoint is unauthenticated, which is why - // replication ships default-OFF. + // configured — nothing dials it, and LocalDbSyncAuthInterceptor (registered on + // AddGrpc above) rejects anything that tries until an ApiKey is configured. app.MapZbLocalDbSync(); // Site-shutdown ordering. ApplicationStopping diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/SiteServiceRegistration.cs b/src/ZB.MOM.WW.ScadaBridge.Host/SiteServiceRegistration.cs index 3abee357..3b2122d7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/SiteServiceRegistration.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/SiteServiceRegistration.cs @@ -100,6 +100,16 @@ public static class SiteServiceRegistration services.AddSiteEventLogHealthMetricsBridge( sp => () => sp.GetRequiredService().FailedWriteCount); + // LocalDb replication — bridge ISyncStatus onto the site health report. Registered + // unconditionally alongside the engine: on a node with no peer this reports + // Connected=false with a real 0 backlog, which is the healthy default-OFF state. + // OplogBacklog is passed through NULLABLE on purpose — null means the poll failed + // or no provider is wired, and flattening it to 0 would render a replication pair + // that cannot read its own oplog as perfectly healthy. + services.AddLocalDbReplicationHealthBridge( + sp => () => sp.GetRequiredService().Connected, + sp => () => sp.GetRequiredService().OplogBacklog); + // Audit Log — site-side hot-path writer + telemetry collaborators. // The SiteAuditTelemetryActor itself is registered by AkkaHostedService // in the site-role block; this call wires every DI dependency it (and diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/LocalDbSyncAuthInterceptorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/LocalDbSyncAuthInterceptorTests.cs new file mode 100644 index 00000000..bf0b3cac --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/LocalDbSyncAuthInterceptorTests.cs @@ -0,0 +1,171 @@ +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.LocalDb.Replication; +using ZB.MOM.WW.ScadaBridge.Host; + +namespace ZB.MOM.WW.ScadaBridge.Host.Tests; + +/// +/// LocalDb Phase 1 (Task 8) — the passive sync endpoint's inbound gate. +/// +/// +/// The replication library's LocalDbSyncService verifies nothing; inbound auth is +/// explicitly the host's job. Without this interceptor, anything able to reach a site +/// node's gRPC port could stream arbitrary rows straight into the consolidated site +/// database — including into OperationTracking, which central reconciles from. +/// +public class LocalDbSyncAuthInterceptorTests +{ + private const string SyncMethod = "/localdb_sync.v1.LocalDbSync/Sync"; + private const string SiteStreamMethod = "/scadabridge.SiteStream/Connect"; + + private static LocalDbSyncAuthInterceptor CreateInterceptor(string? apiKey) + => new( + Options.Create(new ReplicationOptions { ApiKey = apiKey }), + NullLogger.Instance); + + private static ServerCallContext CreateContext(string method, string? authorizationHeader) + { + var headers = new Metadata(); + if (authorizationHeader is not null) + headers.Add("authorization", authorizationHeader); + + return new FakeServerCallContext(method, headers); + } + + /// + /// Minimal carrying just a method name and request + /// headers — the only two things the interceptor reads. + /// + /// + /// Hand-rolled rather than using Grpc.Core.Testing.TestServerCallContext: that + /// type ships in the retired native Grpc.Core package and does not exist on the + /// grpc-dotnet stack this solution runs on. Pulling in the dead package to get one + /// test helper would be a worse trade than these few overrides. + /// + 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; + } + + /// Invokes the interceptor's unary path with a trivial continuation. + private static Task Invoke( + LocalDbSyncAuthInterceptor interceptor, ServerCallContext context) + => interceptor.UnaryServerHandler( + "request", context, (_, _) => Task.FromResult("ok")); + + [Fact] + public async Task NonSyncMethod_PassesThrough_EvenWithNoKeyConfigured() + { + // The interceptor is registered on the shared site AddGrpc pipeline, so it sees + // every call. It must be scoped strictly to the sync service — gating SiteStream + // would break site↔central communication outright. + var interceptor = CreateInterceptor(apiKey: null); + var context = CreateContext(SiteStreamMethod, authorizationHeader: null); + + Assert.Equal("ok", await Invoke(interceptor, context)); + } + + [Fact] + public async Task SyncMethod_WithNoKeyConfigured_IsDenied_EvenWithABearerToken() + { + // Fail-closed. "No key configured" is the DEFAULT every site node ships with, so + // treating it as "no auth required" would silently expose the endpoint on exactly + // the configuration that is most common. Presenting a token must not help. + var interceptor = CreateInterceptor(apiKey: null); + var context = CreateContext(SyncMethod, "Bearer anything-at-all"); + + var ex = await Assert.ThrowsAsync(() => Invoke(interceptor, context)); + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task SyncMethod_WithNoBearerToken_IsDenied() + { + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SyncMethod, authorizationHeader: null); + + var ex = await Assert.ThrowsAsync(() => Invoke(interceptor, context)); + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task SyncMethod_WithWrongBearerToken_IsDenied() + { + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SyncMethod, "Bearer the-wrong-key"); + + var ex = await Assert.ThrowsAsync(() => Invoke(interceptor, context)); + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task SyncMethod_WithCorrectBearerToken_PassesThrough() + { + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SyncMethod, "Bearer the-shared-key"); + + Assert.Equal("ok", await Invoke(interceptor, context)); + } + + [Fact] + public async Task SyncMethod_WithCorrectKey_ButNoBearerScheme_IsDenied() + { + // A raw key with no "Bearer " prefix is not what SyncBackgroundService sends, and + // accepting it would widen the accepted credential shape for no reason. + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SyncMethod, "the-shared-key"); + + var ex = await Assert.ThrowsAsync(() => Invoke(interceptor, context)); + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task SyncMethod_TokenComparison_IsNotAPrefixMatch() + { + // A prefix/StartsWith comparison would accept a truncated key and make the secret + // recoverable one character at a time. + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SyncMethod, "Bearer the-shared-ke"); + + var ex = await Assert.ThrowsAsync(() => Invoke(interceptor, context)); + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } + + [Fact] + public async Task DuplexStreaming_IsGated_BecauseThatIsHowSyncActuallyRuns() + { + // The sync RPC is a bidirectional stream. Gating only the unary path would leave + // the real endpoint wide open while every unary test still passed. + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SyncMethod, "Bearer the-wrong-key"); + + var ex = await Assert.ThrowsAsync(() => + interceptor.DuplexStreamingServerHandler( + requestStream: null!, + responseStream: null!, + context, + (_, _, _) => Task.CompletedTask)); + + Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbWiringTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbWiringTests.cs index db34fa6b..cff13488 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbWiringTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbWiringTests.cs @@ -1,6 +1,9 @@ using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; using ZB.MOM.WW.LocalDb; using ZB.MOM.WW.LocalDb.Replication; using ZB.MOM.WW.ScadaBridge.Host; @@ -150,6 +153,49 @@ public class SiteLocalDbWiringTests : IDisposable Assert.Equal(0L, status.OplogBacklog); } + [Fact] + public void Site_Replication_HealthBridge_IsRegisteredAsAHostedService() + { + // Task 9. Registered via a factory lambda, so ImplementationType is null and the + // only honest assertion is on the resolved instance. + var hosted = _host.Services.GetServices(); + + Assert.Contains(hosted, h => h is LocalDbReplicationStatusReporter); + } + + [Fact] + public void Site_Replication_HealthBridge_PublishesStatusOntoTheHealthReport() + { + // End-to-end through the REAL collector: probe once, then collect. This is what + // catches a bridge that is registered but wired to the wrong provider — the + // registration test above would pass either way. + var reporter = _host.Services.GetServices() + .OfType() + .Single(); + var collector = _host.Services.GetRequiredService(); + + reporter.Probe(); + var report = collector.CollectReport("TestSite"); + + // No peer configured: not connected, and a REAL 0 backlog rather than null. + Assert.False(report.LocalDbReplicationConnected); + Assert.Equal(0L, report.LocalDbOplogBacklog); + } + + [Fact] + public void Site_HealthReport_LocalDbFields_AreNull_BeforeTheReporterHasRun() + { + // Null means "no data yet", and must be distinguishable from a real + // "disconnected, zero backlog". A collector that defaulted these to false/0 would + // report an unwired node as a healthy connected one. + var collector = new SiteHealthCollector(); + + var report = collector.CollectReport("TestSite"); + + Assert.Null(report.LocalDbReplicationConnected); + Assert.Null(report.LocalDbOplogBacklog); + } + [Fact] public void Site_LocalDb_CreatesTheConfiguredFile() {