diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/TelemetryStreamAuthInterceptor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/TelemetryStreamAuthInterceptor.cs new file mode 100644 index 00000000..7f331c7b --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/TelemetryStreamAuthInterceptor.cs @@ -0,0 +1,165 @@ +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.OtOpcUa.Cluster; + +namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; + +/// +/// Gates the driver-node telemetry stream endpoint (per-cluster mesh Phase 5). The +/// TelemetryStreamService.Subscribe RPC streams a node's live alerts / script-logs / +/// driver-health / driver-resilience-status to whoever dials it; anything able to reach a driver +/// node's telemetry port could otherwise pull that live feed out-of-band. This interceptor is the +/// ONLY inbound auth on that endpoint. +/// +/// +/// +/// Scoped by method path. Only calls under +/// /telemetry.v1.TelemetryStreamService/ are gated; every other method on the shared +/// gRPC pipeline (config-serve, LocalDb sync) passes through untouched, so this can share one +/// AddGrpc registration with and +/// . +/// +/// +/// Fail-closed. With no Telemetry:ApiKey configured, NO dial is accepted, +/// authenticated or not. Treating "no key" as "no auth required" would expose the live +/// telemetry feed on exactly the default shape most nodes ship with. The caller dialing in +/// must present the same TelemetryDial:ApiKey (the shared node key), so a key typo +/// stops telemetry delivery outright rather than degrading to unauthenticated. +/// +/// +/// 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. +/// +/// +/// Exactly one public constructor. Grpc.AspNetCore silently stops invoking an +/// interceptor that has more than one public ctor — see +/// TelemetryStreamAuthInterceptorTests.HasExactlyOnePublicConstructor. +/// +/// +public sealed class TelemetryStreamAuthInterceptor : Interceptor +{ + private const string ServicePrefix = "/telemetry.v1.TelemetryStreamService/"; + private const string AuthorizationHeader = "authorization"; + private const string BearerPrefix = "Bearer "; + + private readonly IOptions _options; + private readonly ILogger _logger; + + /// Creates the interceptor. + /// Telemetry options; ApiKey is the expected bearer token. + /// Logger for denial diagnostics. + public TelemetryStreamAuthInterceptor( + 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 telemetry-stream call that does not carry the configured bearer token. Non-telemetry + /// 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 telemetry stream call to {Method}: no Telemetry:ApiKey is configured, so " + + "the telemetry endpoint is closed. Configure the shared node key on this node and every " + + "dialing caller.", + context.Method); + throw new RpcException(new Status( + StatusCode.PermissionDenied, + "Telemetry stream 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 telemetry stream 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, + "Telemetry stream 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/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/TelemetryStreamAuthInterceptorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/TelemetryStreamAuthInterceptorTests.cs new file mode 100644 index 00000000..287945d7 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/TelemetryStreamAuthInterceptorTests.cs @@ -0,0 +1,199 @@ +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Host.Configuration; + +namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Configuration; + +/// +/// Per-cluster mesh Phase 5 — the telemetry stream endpoint's inbound gate. Wiring this +/// interceptor into the AddGrpc pipeline + Kestrel is a separate, later task; this test +/// covers only the interceptor's own authorization behavior. +/// +/// +/// Same contract as / , +/// scoped to /telemetry.v1.TelemetryStreamService/ — fail-closed on an unconfigured key, +/// constant-time comparison, gated on every handler shape so scoping holds regardless of RPC shape. +/// +public sealed class TelemetryStreamAuthInterceptorTests +{ + private const string SubscribeMethod = "/telemetry.v1.TelemetryStreamService/Subscribe"; + private const string OtherMethod = "/deployment_artifact.v1.DeploymentArtifactService/Fetch"; + + private static TelemetryStreamAuthInterceptor CreateInterceptor(string? apiKey) + => new( + Options.Create(new TelemetryOptions { ApiKey = apiKey ?? string.Empty }), + 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); + } + + /// Invokes the interceptor's server-streaming path (how Subscribe actually runs). + private static Task Invoke(TelemetryStreamAuthInterceptor interceptor, ServerCallContext context) + => interceptor.ServerStreamingServerHandler( + "request", responseStream: null!, context, (_, _, _) => Task.CompletedTask); + + [Fact] + public async Task NonTelemetryMethod_PassesThrough_EvenWithNoKeyConfigured() + { + // The interceptor shares the AddGrpc pipeline with other gRPC services, so it sees every + // call. It must be scoped strictly to the telemetry service. + var interceptor = CreateInterceptor(apiKey: null); + var context = CreateContext(OtherMethod, authorizationHeader: null); + + await Invoke(interceptor, context); // no throw + } + + [Fact] + public async Task SubscribeMethod_WithNoKeyConfigured_IsDenied_EvenWithABearerToken() + { + // Fail-closed. "No key configured" is the DEFAULT — treating it as "no auth required" would + // expose live telemetry on exactly the shape most nodes ship with. A token must not help. + var interceptor = CreateInterceptor(apiKey: null); + var context = CreateContext(SubscribeMethod, "Bearer anything-at-all"); + + var ex = await Should.ThrowAsync(() => Invoke(interceptor, context)); + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task SubscribeMethod_WithNoBearerToken_IsDenied() + { + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SubscribeMethod, authorizationHeader: null); + + var ex = await Should.ThrowAsync(() => Invoke(interceptor, context)); + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task SubscribeMethod_WithMalformedAuthorizationHeader_IsDenied() + { + // No "Bearer " prefix at all — must be treated the same as a missing header. + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SubscribeMethod, "the-shared-key"); + + var ex = await Should.ThrowAsync(() => Invoke(interceptor, context)); + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task SubscribeMethod_WithWrongBearerToken_IsDenied() + { + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SubscribeMethod, "Bearer the-wrong-key"); + + var ex = await Should.ThrowAsync(() => Invoke(interceptor, context)); + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task SubscribeMethod_WithCorrectBearerToken_PassesThrough() + { + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SubscribeMethod, "Bearer the-shared-key"); + + await Invoke(interceptor, context); // no throw + } + + [Fact] + public async Task SubscribeMethod_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(SubscribeMethod, "Bearer the-shared-ke"); + + var ex = await Should.ThrowAsync(() => Invoke(interceptor, context)); + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task UnaryPath_IsAlsoGated() + { + // Subscribe is server-streaming today, but gating only that path would leave any future + // unary method on the same service open. Gate every handler shape. + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SubscribeMethod, "Bearer the-wrong-key"); + + var ex = await Should.ThrowAsync(() => + interceptor.UnaryServerHandler( + "request", context, (_, _) => Task.FromResult("ok"))); + + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task ClientStreamingPath_IsAlsoGated() + { + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SubscribeMethod, "Bearer the-wrong-key"); + + var ex = await Should.ThrowAsync(() => + interceptor.ClientStreamingServerHandler( + requestStream: null!, context, (_, _) => Task.FromResult("ok"))); + + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task DuplexStreamingPath_IsAlsoGated() + { + var interceptor = CreateInterceptor("the-shared-key"); + var context = CreateContext(SubscribeMethod, "Bearer the-wrong-key"); + + var ex = await Should.ThrowAsync(() => + interceptor.DuplexStreamingServerHandler( + requestStream: null!, responseStream: null!, context, (_, _, _) => Task.CompletedTask)); + + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public void HasExactlyOnePublicConstructor() + { + // Grpc.AspNetCore silently stops invoking an interceptor with more than one public ctor — + // this pins the shape so a well-meaning overload doesn't quietly disable auth in production. + var publicCtors = typeof(TelemetryStreamAuthInterceptor).GetConstructors(); + + publicCtors.Length.ShouldBe(1); + } + + /// + /// Minimal carrying just a method name and request headers — + /// the only two things the interceptor reads. Hand-rolled because + /// Grpc.Core.Testing.TestServerCallContext ships only in the retired native package. + /// + 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; + } +}