From 19cc74f21a4e974fc5e676901e21d2705b71d3ab Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:14:33 -0400 Subject: [PATCH] fix(auth): authenticate client- and duplex-streaming gRPC calls GatewayGrpcAuthorizationInterceptor only overrode the unary and server-streaming handlers; the base Interceptor passes client- and duplex-streaming calls through unauthenticated. Both now run the same API-key pipeline and fail closed to the admin scope, since there is no single request message to resolve a scope from. --- docs/Authorization.md | 9 +- .../GatewayGrpcAuthorizationInterceptor.cs | 57 ++++++- ...atewayGrpcAuthorizationInterceptorTests.cs | 156 ++++++++++++++++++ 3 files changed, 211 insertions(+), 11 deletions(-) diff --git a/docs/Authorization.md b/docs/Authorization.md index ed5ec8b..23c71d1 100644 --- a/docs/Authorization.md +++ b/docs/Authorization.md @@ -10,7 +10,7 @@ Authorization runs as a single gRPC server interceptor registered for every call The participating types live under `src/ZB.MOM.WW.MxGateway.Server/Security/Authorization/`: -- `GatewayGrpcAuthorizationInterceptor` runs the authenticate-then-authorize pipeline for unary and server-streaming calls. +- `GatewayGrpcAuthorizationInterceptor` runs the authenticate-then-authorize pipeline for every call shape: unary, server-streaming, client-streaming, and duplex-streaming. - `GatewayGrpcScopeResolver` maps a request message (and, for `MxCommandRequest`, the inner `MxCommandKind`) to the scope string that must be present on the caller. - `GatewayScopes` exposes the canonical scope constants used by the resolver and any downstream consumer. - `GatewayRequestIdentityAccessor` and `IGatewayRequestIdentityAccessor` expose the verified identity to handlers and any service code that runs inside the call. @@ -32,7 +32,7 @@ Centralizing the policy in `GatewayGrpcAuthorizationInterceptor` produces three ## Interceptor Flow -`GatewayGrpcAuthorizationInterceptor` overrides both `UnaryServerHandler` and `ServerStreamingServerHandler`. Both call the same private `AuthenticateAndAuthorizeAsync` helper before invoking the continuation, then push the resolved identity onto the accessor for the duration of the call. +`GatewayGrpcAuthorizationInterceptor` overrides all four server handlers: `UnaryServerHandler`, `ServerStreamingServerHandler`, `ClientStreamingServerHandler`, and `DuplexStreamingServerHandler`. Each calls the same private `AuthenticateAndAuthorizeAsync` helper before invoking the continuation, then pushes the resolved identity onto the accessor for the duration of the call. The base `Interceptor` passes any handler it does not override straight through, so a client- or duplex-streaming RPC added to a mapped service would otherwise run with no API key check at all. The unary and server-streaming handlers resolve the required scope from the request message; the client- and duplex-streaming handlers have no single request to inspect before the handler reads the stream, so they fail closed to `admin` — the same scope the resolver assigns an unrecognized request type. No public RPC uses either shape today. ```csharp public override async Task UnaryServerHandler( @@ -40,7 +40,10 @@ public override async Task UnaryServerHandler( ServerCallContext context, UnaryServerMethod continuation) { - ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync(request, context).ConfigureAwait(false); + ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync( + scopeResolver.ResolveRequiredScope(request), + context) + .ConfigureAwait(false); IDisposable? identityScope = identity is null ? null : identityAccessor.Push(identity); using (identityScope) { diff --git a/src/ZB.MOM.WW.MxGateway.Server/Security/Authorization/GatewayGrpcAuthorizationInterceptor.cs b/src/ZB.MOM.WW.MxGateway.Server/Security/Authorization/GatewayGrpcAuthorizationInterceptor.cs index 345ed4d..730c2e2 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Security/Authorization/GatewayGrpcAuthorizationInterceptor.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Security/Authorization/GatewayGrpcAuthorizationInterceptor.cs @@ -30,7 +30,10 @@ public sealed class GatewayGrpcAuthorizationInterceptor( ServerCallContext context, UnaryServerMethod continuation) { - ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync(request, context).ConfigureAwait(false); + ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync( + scopeResolver.ResolveRequiredScope(request), + context) + .ConfigureAwait(false); IDisposable? identityScope = identity is null ? null : identityAccessor.Push(identity); using (identityScope) { @@ -45,7 +48,10 @@ public sealed class GatewayGrpcAuthorizationInterceptor( ServerCallContext context, ServerStreamingServerMethod continuation) { - ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync(request, context).ConfigureAwait(false); + ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync( + scopeResolver.ResolveRequiredScope(request), + context) + .ConfigureAwait(false); IDisposable? identityScope = identity is null ? null : identityAccessor.Push(identity); using (identityScope) { @@ -53,15 +59,51 @@ public sealed class GatewayGrpcAuthorizationInterceptor( } } + // The base Interceptor passes client- and duplex-streaming calls straight to the handler, so + // without these overrides any such RPC (none exist today, but a .proto edit could add one, and + // the shared Galaxy Repository service is mapped through the same interceptor) would be + // reachable with no API key at all. Both run the identical authentication path. There is no + // single request message to derive a scope from before the handler reads the stream, so they + // fail closed to the scope the resolver assigns an unrecognized request: admin. + + /// + public override async Task ClientStreamingServerHandler( + IAsyncStreamReader requestStream, + ServerCallContext context, + ClientStreamingServerMethod continuation) + { + ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync(GatewayScopes.Admin, context) + .ConfigureAwait(false); + IDisposable? identityScope = identity is null ? null : identityAccessor.Push(identity); + using (identityScope) + { + return await continuation(requestStream, context).ConfigureAwait(false); + } + } + + /// + public override async Task DuplexStreamingServerHandler( + IAsyncStreamReader requestStream, + IServerStreamWriter responseStream, + ServerCallContext context, + DuplexStreamingServerMethod continuation) + { + ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync(GatewayScopes.Admin, context) + .ConfigureAwait(false); + IDisposable? identityScope = identity is null ? null : identityAccessor.Push(identity); + using (identityScope) + { + await continuation(requestStream, responseStream, context).ConfigureAwait(false); + } + } + /// Authenticates the API key and authorizes the RPC call by required scope. - /// Request message type. - /// Request payload. + /// The scope the call requires, resolved by the caller. /// RPC server call context. /// Authenticated API key identity, or null if authentication is disabled. - private async Task AuthenticateAndAuthorizeAsync( - TRequest request, + private async Task AuthenticateAndAuthorizeAsync( + string requiredScope, ServerCallContext context) - where TRequest : class { if (options.Value.Authentication.Mode == AuthenticationMode.Disabled) { @@ -116,7 +158,6 @@ public sealed class GatewayGrpcAuthorizationInterceptor( ApiKeyIdentity identity = GatewayApiKeyIdentityMapper.ToGatewayIdentity(verification.Identity); - string requiredScope = scopeResolver.ResolveRequiredScope(request); if (!identity.Scopes.Contains(requiredScope)) { throw new RpcException(new Status( diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs index c38b7db..76fc2ab 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs @@ -717,6 +717,151 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests Assert.Null(GatewayGrpcAuthorizationInterceptor.TryResolveKeyId($"Bearer mxgw_{overCap}_secret")); } + /// + /// A client-streaming call is authenticated exactly like a unary call: a missing key is + /// rejected with Unauthenticated and the handler never runs. The base Interceptor passes + /// client- and duplex-streaming calls straight through, so without the override any such RPC + /// added to the service would be reachable unauthenticated. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ClientStreamingServerHandler_MissingApiKey_ReturnsUnauthenticatedWithoutRunningHandler() + { + GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor( + new FakeApiKeyVerifier(Failure(ApiKeyFailure.MissingOrMalformed)), + new GatewayRequestIdentityAccessor()); + bool handlerRan = false; + + RpcException exception = await Assert.ThrowsAsync( + () => interceptor.ClientStreamingServerHandler( + new EmptyStreamReader(), + new TestServerCallContext([]), + (_, _) => + { + handlerRan = true; + return Task.FromResult(new MxCommandReply()); + })); + + Assert.Equal(StatusCode.Unauthenticated, exception.StatusCode); + Assert.False(handlerRan); + } + + /// + /// A client-streaming call has no single request message to derive a scope from, so the + /// interceptor fails closed to the admin scope, the resolver's default for an + /// unrecognized request. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ClientStreamingServerHandler_KeyWithoutAdminScope_ReturnsPermissionDenied() + { + GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor( + new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.InvokeRead, GatewayScopes.InvokeWrite)), + new GatewayRequestIdentityAccessor()); + + RpcException exception = await Assert.ThrowsAsync( + () => interceptor.ClientStreamingServerHandler( + new EmptyStreamReader(), + ContextWithAuthorization("Bearer mxgw_operator01_secret"), + (_, _) => Task.FromResult(new MxCommandReply()))); + + Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode); + Assert.Contains(GatewayScopes.Admin, exception.Status.Detail, StringComparison.Ordinal); + } + + /// An admin-scoped key reaches the client-streaming handler with its identity pushed. + /// A task that represents the asynchronous operation. + [Fact] + public async Task ClientStreamingServerHandler_AdminScope_RunsHandlerWithIdentity() + { + GatewayRequestIdentityAccessor identityAccessor = new(); + GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor( + new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.Admin)), + identityAccessor); + string? keyIdSeenByHandler = null; + + await interceptor.ClientStreamingServerHandler( + new EmptyStreamReader(), + ContextWithAuthorization("Bearer mxgw_operator01_secret"), + (_, _) => + { + keyIdSeenByHandler = identityAccessor.Current?.KeyId; + return Task.FromResult(new MxCommandReply()); + }); + + Assert.Equal("operator01", keyIdSeenByHandler); + Assert.Null(identityAccessor.Current); + } + + /// A duplex-streaming call with no key is rejected before the handler runs. + /// A task that represents the asynchronous operation. + [Fact] + public async Task DuplexStreamingServerHandler_MissingApiKey_ReturnsUnauthenticatedWithoutRunningHandler() + { + GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor( + new FakeApiKeyVerifier(Failure(ApiKeyFailure.MissingOrMalformed)), + new GatewayRequestIdentityAccessor()); + bool handlerRan = false; + + RpcException exception = await Assert.ThrowsAsync( + () => interceptor.DuplexStreamingServerHandler( + new EmptyStreamReader(), + new RecordingServerStreamWriter(), + new TestServerCallContext([]), + (_, _, _) => + { + handlerRan = true; + return Task.CompletedTask; + })); + + Assert.Equal(StatusCode.Unauthenticated, exception.StatusCode); + Assert.False(handlerRan); + } + + /// A duplex-streaming call fails closed to the admin scope, like client streaming. + /// A task that represents the asynchronous operation. + [Fact] + public async Task DuplexStreamingServerHandler_KeyWithoutAdminScope_ReturnsPermissionDenied() + { + GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor( + new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.EventsRead)), + new GatewayRequestIdentityAccessor()); + + RpcException exception = await Assert.ThrowsAsync( + () => interceptor.DuplexStreamingServerHandler( + new EmptyStreamReader(), + new RecordingServerStreamWriter(), + ContextWithAuthorization("Bearer mxgw_operator01_secret"), + (_, _, _) => Task.CompletedTask)); + + Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode); + } + + /// An admin-scoped key reaches the duplex handler with its identity pushed. + /// A task that represents the asynchronous operation. + [Fact] + public async Task DuplexStreamingServerHandler_AdminScope_RunsHandlerWithIdentity() + { + GatewayRequestIdentityAccessor identityAccessor = new(); + GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor( + new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.Admin)), + identityAccessor); + RecordingServerStreamWriter writer = new(); + + await interceptor.DuplexStreamingServerHandler( + new EmptyStreamReader(), + writer, + ContextWithAuthorization("Bearer mxgw_operator01_secret"), + async (_, responseStream, _) => + { + Assert.Equal("operator01", identityAccessor.Current?.KeyId); + await responseStream.WriteAsync(new MxCommandReply { SessionId = "duplex" }); + }); + + Assert.Equal("duplex", Assert.Single(writer.Messages).SessionId); + Assert.Null(identityAccessor.Current); + } + private static MxAccessGatewayService CreateService( ISessionManager sessionManager, IGatewayRequestIdentityAccessor identityAccessor) @@ -933,4 +1078,15 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests } } + + /// An that yields no messages. + /// The streamed message type. + private sealed class EmptyStreamReader : IAsyncStreamReader + { + /// + public T Current => throw new InvalidOperationException("The stream is empty."); + + /// + public Task MoveNext(CancellationToken cancellationToken) => Task.FromResult(false); + } }