diff --git a/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/ForwardedAuthenticationTests.cs b/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/ForwardedAuthenticationTests.cs new file mode 100644 index 000000000..120f621d9 --- /dev/null +++ b/src/EventStore.Core.Tests/Services/Transport/Grpc/StreamsTests/ForwardedAuthenticationTests.cs @@ -0,0 +1,141 @@ +using System; +using System.Collections.Generic; +using System.Reflection; +using System.Security.Claims; +using System.Threading; +using System.Threading.Tasks; +using EventStore.Client.Streams; +using EventStore.Core.Authorization; +using EventStore.Core.Bus; +using EventStore.Core.Messages; +using EventStore.Core.Messaging; +using EventStore.Core.Services.Transport.Grpc; +using Google.Protobuf; +using Grpc.Core; +using Microsoft.AspNetCore.Http; +using NUnit.Framework; +using GrpcMetadata = EventStore.Core.Services.Transport.Grpc.Constants.Metadata; +using GrpcStreams = EventStore.Client.Streams.Streams; + +namespace EventStore.Core.Tests.Services.Transport.Grpc.StreamsTests; + +[TestFixture] +public class ForwardedAuthenticationTests +{ + [Test] + public void append_reports_forwarded_authentication_failure() + { + var service = CreateService(new AuthenticationFailurePublisher()); + var requests = new EnumerableStreamReader([ + new AppendReq { + Options = new AppendReq.Types.Options { + NoStream = new(), + StreamIdentifier = "forwarded-auth-append" + } + }, + new AppendReq { + ProposedMessage = new AppendReq.Types.ProposedMessage { + Id = Uuid.NewUuid().ToDto(), + Metadata = { + [GrpcMetadata.Type] = "test", + [GrpcMetadata.ContentType] = GrpcMetadata.ContentTypes.ApplicationJson + }, + Data = ByteString.CopyFromUtf8("{}") + } + } + ]); + + var exception = Assert.ThrowsAsync(() => service.Append(requests, new TestServerCallContext())); + + Assert.That(exception!.StatusCode, Is.EqualTo(StatusCode.Unauthenticated)); + Assert.That(exception.Status.Detail, Does.Contain("forwarding denied")); + } + + [Test] + public void delete_reports_forwarded_authentication_failure() + { + var service = CreateService(new AuthenticationFailurePublisher()); + var request = new DeleteReq + { + Options = new DeleteReq.Types.Options + { + NoStream = new(), + StreamIdentifier = "forwarded-auth-delete" + } + }; + + var exception = Assert.ThrowsAsync(() => service.Delete(request, new TestServerCallContext())); + + Assert.That(exception!.StatusCode, Is.EqualTo(StatusCode.Unauthenticated)); + Assert.That(exception.Status.Detail, Does.Contain("forwarding denied")); + } + + private static GrpcStreams.StreamsBase CreateService(IPublisher publisher) + { + var type = typeof(GrpcTrackers).Assembly.GetType( + "EventStore.Core.Services.Transport.Grpc.Streams`1", throwOnError: true)! + .MakeGenericType(typeof(string)); + return (GrpcStreams.StreamsBase)Activator.CreateInstance(type, + BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic, + binder: null, + args: [publisher, 1024, TimeSpan.FromSeconds(1), null, new GrpcTrackers(), new PassthroughAuthorizationProvider()], + culture: null)!; + } + + private sealed class AuthenticationFailurePublisher : IPublisher + { + public void Publish(Message message) + { + if (message is not ClientMessage.WriteRequestMessage request) + { + throw new InvalidOperationException($"Unexpected message {message.GetType().Name}"); + } + + request.Envelope.ReplyWith(new ClientMessage.NotAuthenticated(request.CorrelationId, "forwarding denied")); + } + } + + private sealed class EnumerableStreamReader(IEnumerable values) : IAsyncStreamReader + { + private readonly IEnumerator _values = values.GetEnumerator(); + public T Current { get; private set; } = default!; + + public Task MoveNext(CancellationToken cancellationToken) + { + if (!_values.MoveNext()) + { + return Task.FromResult(false); + } + + Current = _values.Current; + return Task.FromResult(true); + } + } + + private sealed class TestServerCallContext : ServerCallContext + { + public TestServerCallContext() + { + UserStateCore["__HttpContext"] = new DefaultHttpContext + { + User = new ClaimsPrincipal(new ClaimsIdentity()) + }; + } + + protected override string MethodCore => "/event_store.client.streams.Streams/Append"; + protected override string HostCore => "localhost"; + protected override string PeerCore => "ipv4:127.0.0.1:2113"; + protected override DateTime DeadlineCore => DateTime.MaxValue; + protected override Metadata RequestHeadersCore { get; } = new(); + protected override CancellationToken CancellationTokenCore => CancellationToken.None; + protected override Metadata ResponseTrailersCore { get; } = new(); + protected override Status StatusCore { get; set; } + protected override WriteOptions WriteOptionsCore { get; set; } + protected override AuthContext AuthContextCore { get; } = + new(string.Empty, new Dictionary>()); + protected override IDictionary UserStateCore { get; } = new Dictionary(); + protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask; + protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions options) => + throw new NotSupportedException(); + } +} diff --git a/src/EventStore.Core/Services/Transport/Grpc/Streams.Append.cs b/src/EventStore.Core/Services/Transport/Grpc/Streams.Append.cs index ddff83de3..3cc40c215 100644 --- a/src/EventStore.Core/Services/Transport/Grpc/Streams.Append.cs +++ b/src/EventStore.Core/Services/Transport/Grpc/Streams.Append.cs @@ -122,6 +122,12 @@ void HandleWriteEventsCompleted(Message message) appendResponseSource.TrySetException(ex); return; } + if (message is ClientMessage.NotAuthenticated notAuthenticated) + { + appendResponseSource.TrySetException(new RpcException( + new Status(StatusCode.Unauthenticated, notAuthenticated.Reason))); + return; + } if (!(message is ClientMessage.WriteEventsCompleted completed)) { diff --git a/src/EventStore.Core/Services/Transport/Grpc/Streams.Delete.cs b/src/EventStore.Core/Services/Transport/Grpc/Streams.Delete.cs index c798b152c..bd0a441ed 100644 --- a/src/EventStore.Core/Services/Transport/Grpc/Streams.Delete.cs +++ b/src/EventStore.Core/Services/Transport/Grpc/Streams.Delete.cs @@ -148,6 +148,12 @@ void HandleStreamDeletedCompleted(Message message) deleteResponseSource.TrySetException(ex); return; } + if (message is ClientMessage.NotAuthenticated notAuthenticated) + { + deleteResponseSource.TrySetException(new RpcException( + new Status(StatusCode.Unauthenticated, notAuthenticated.Reason))); + return; + } if (message is not ClientMessage.DeleteStreamCompleted completed) {