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.
This commit is contained in:
Joseph Doherty
2026-10-22 06:14:33 -04:00
parent ca0cf19a1f
commit 19cc74f21a
3 changed files with 211 additions and 11 deletions
+6 -3
View File
@@ -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<TResponse> UnaryServerHandler<TRequest, TResponse>(
@@ -40,7 +40,10 @@ public override async Task<TResponse> UnaryServerHandler<TRequest, TResponse>(
ServerCallContext context,
UnaryServerMethod<TRequest, TResponse> 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)
{
@@ -30,7 +30,10 @@ public sealed class GatewayGrpcAuthorizationInterceptor(
ServerCallContext context,
UnaryServerMethod<TRequest, TResponse> 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<TRequest, TResponse> 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.
/// <inheritdoc />
public override async Task<TResponse> ClientStreamingServerHandler<TRequest, TResponse>(
IAsyncStreamReader<TRequest> requestStream,
ServerCallContext context,
ClientStreamingServerMethod<TRequest, TResponse> 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);
}
}
/// <inheritdoc />
public override async Task DuplexStreamingServerHandler<TRequest, TResponse>(
IAsyncStreamReader<TRequest> requestStream,
IServerStreamWriter<TResponse> responseStream,
ServerCallContext context,
DuplexStreamingServerMethod<TRequest, TResponse> 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);
}
}
/// <summary>Authenticates the API key and authorizes the RPC call by required scope.</summary>
/// <typeparam name="TRequest">Request message type.</typeparam>
/// <param name="request">Request payload.</param>
/// <param name="requiredScope">The scope the call requires, resolved by the caller.</param>
/// <param name="context">RPC server call context.</param>
/// <returns>Authenticated API key identity, or null if authentication is disabled.</returns>
private async Task<ApiKeyIdentity?> AuthenticateAndAuthorizeAsync<TRequest>(
TRequest request,
private async Task<ApiKeyIdentity?> 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(
@@ -717,6 +717,151 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests
Assert.Null(GatewayGrpcAuthorizationInterceptor.TryResolveKeyId($"Bearer mxgw_{overCap}_secret"));
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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<RpcException>(
() => interceptor.ClientStreamingServerHandler<MxCommandRequest, MxCommandReply>(
new EmptyStreamReader<MxCommandRequest>(),
new TestServerCallContext([]),
(_, _) =>
{
handlerRan = true;
return Task.FromResult(new MxCommandReply());
}));
Assert.Equal(StatusCode.Unauthenticated, exception.StatusCode);
Assert.False(handlerRan);
}
/// <summary>
/// A client-streaming call has no single request message to derive a scope from, so the
/// interceptor fails closed to the <c>admin</c> scope, the resolver's default for an
/// unrecognized request.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task ClientStreamingServerHandler_KeyWithoutAdminScope_ReturnsPermissionDenied()
{
GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.InvokeRead, GatewayScopes.InvokeWrite)),
new GatewayRequestIdentityAccessor());
RpcException exception = await Assert.ThrowsAsync<RpcException>(
() => interceptor.ClientStreamingServerHandler<MxCommandRequest, MxCommandReply>(
new EmptyStreamReader<MxCommandRequest>(),
ContextWithAuthorization("Bearer mxgw_operator01_secret"),
(_, _) => Task.FromResult(new MxCommandReply())));
Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode);
Assert.Contains(GatewayScopes.Admin, exception.Status.Detail, StringComparison.Ordinal);
}
/// <summary>An admin-scoped key reaches the client-streaming handler with its identity pushed.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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<MxCommandRequest, MxCommandReply>(
new EmptyStreamReader<MxCommandRequest>(),
ContextWithAuthorization("Bearer mxgw_operator01_secret"),
(_, _) =>
{
keyIdSeenByHandler = identityAccessor.Current?.KeyId;
return Task.FromResult(new MxCommandReply());
});
Assert.Equal("operator01", keyIdSeenByHandler);
Assert.Null(identityAccessor.Current);
}
/// <summary>A duplex-streaming call with no key is rejected before the handler runs.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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<RpcException>(
() => interceptor.DuplexStreamingServerHandler<MxCommandRequest, MxCommandReply>(
new EmptyStreamReader<MxCommandRequest>(),
new RecordingServerStreamWriter<MxCommandReply>(),
new TestServerCallContext([]),
(_, _, _) =>
{
handlerRan = true;
return Task.CompletedTask;
}));
Assert.Equal(StatusCode.Unauthenticated, exception.StatusCode);
Assert.False(handlerRan);
}
/// <summary>A duplex-streaming call fails closed to the admin scope, like client streaming.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task DuplexStreamingServerHandler_KeyWithoutAdminScope_ReturnsPermissionDenied()
{
GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.EventsRead)),
new GatewayRequestIdentityAccessor());
RpcException exception = await Assert.ThrowsAsync<RpcException>(
() => interceptor.DuplexStreamingServerHandler<MxCommandRequest, MxCommandReply>(
new EmptyStreamReader<MxCommandRequest>(),
new RecordingServerStreamWriter<MxCommandReply>(),
ContextWithAuthorization("Bearer mxgw_operator01_secret"),
(_, _, _) => Task.CompletedTask));
Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode);
}
/// <summary>An admin-scoped key reaches the duplex handler with its identity pushed.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task DuplexStreamingServerHandler_AdminScope_RunsHandlerWithIdentity()
{
GatewayRequestIdentityAccessor identityAccessor = new();
GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.Admin)),
identityAccessor);
RecordingServerStreamWriter<MxCommandReply> writer = new();
await interceptor.DuplexStreamingServerHandler<MxCommandRequest, MxCommandReply>(
new EmptyStreamReader<MxCommandRequest>(),
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
}
}
/// <summary>An <see cref="IAsyncStreamReader{T}"/> that yields no messages.</summary>
/// <typeparam name="T">The streamed message type.</typeparam>
private sealed class EmptyStreamReader<T> : IAsyncStreamReader<T>
{
/// <inheritdoc />
public T Current => throw new InvalidOperationException("The stream is empty.");
/// <inheritdoc />
public Task<bool> MoveNext(CancellationToken cancellationToken) => Task.FromResult(false);
}
}