Merge 616cb81 into 398145f

Author: Gsantomaggio

RabbitMQ.Stream.Client/Client.cs

@@ -18,6 +18,12 @@
 
 namespace RabbitMQ.Stream.Client
 {
+    public enum AuthMechanism
+    {
+        Plain,
+        External,
+    }
+
     public record ClientParameters
     {
         // internal list of endpoints where the client will try to connect
@@ -63,6 +69,8 @@ public string ClientProvidedName
         public SslOption Ssl { get; set; } = new SslOption();
 
         public AddressResolver AddressResolver { get; set; } = null;
+
+        public AuthMechanism AuthMechanism { get; set; } = AuthMechanism.Plain;
     }
 
     internal readonly struct OutgoingMsg : ICommand
@@ -214,11 +222,20 @@ await client
                     .ConfigureAwait(false);
             logger?.LogDebug("Sasl mechanism: {Mechanisms}", saslHandshakeResponse.Mechanisms);
 
+            var isValid = saslHandshakeResponse.Mechanisms.Contains(parameters.AuthMechanism.ToString().ToUpper(),
+                StringComparer.OrdinalIgnoreCase);
+            if (!isValid)
+            {
+                throw new AuthMechanismNotSupportedException(
+                    $"Sasl mechanism {parameters.AuthMechanism} is not supported by the server");
+            }
+
             var saslData = Encoding.UTF8.GetBytes($"\0{parameters.UserName}\0{parameters.Password}");
             var authResponse =
                 await client
                     .Request<SaslAuthenticateRequest, SaslAuthenticateResponse>(corr =>
-                        new SaslAuthenticateRequest(corr, "PLAIN", saslData)).ConfigureAwait(false);
+                        new SaslAuthenticateRequest(corr, parameters.AuthMechanism.ToString().ToUpper(), saslData))
+                    .ConfigureAwait(false);
             ClientExceptions.MaybeThrowException(authResponse.ResponseCode, parameters.UserName);
 
             //tune

RabbitMQ.Stream.Client/ClientExceptions.cs

@@ -93,4 +93,12 @@ public RouteNotFoundException(string s)
         {
         }
     }
+
+    public class AuthMechanismNotSupportedException : Exception
+    {
+        public AuthMechanismNotSupportedException(string s)
+            : base(s)
+        {
+        }
+    }
 }

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -2,10 +2,17 @@ const RabbitMQ.Stream.Client.RouteQueryResponse.Key = 24 -> ushort
 const RabbitMQ.Stream.Client.StreamStatsResponse.Key = 28 -> ushort
 RabbitMQ.Stream.Client.AbstractEntity.MaybeCancelToken() -> void
 RabbitMQ.Stream.Client.AbstractEntity.Token.get -> System.Threading.CancellationToken
+RabbitMQ.Stream.Client.AuthMechanism
+RabbitMQ.Stream.Client.AuthMechanism.External = 1 -> RabbitMQ.Stream.Client.AuthMechanism
+RabbitMQ.Stream.Client.AuthMechanism.Plain = 0 -> RabbitMQ.Stream.Client.AuthMechanism
+RabbitMQ.Stream.Client.AuthMechanismNotSupportedException
+RabbitMQ.Stream.Client.AuthMechanismNotSupportedException.AuthMechanismNotSupportedException(string s) -> void
 RabbitMQ.Stream.Client.Chunk.Data.get -> System.Memory<byte>
 RabbitMQ.Stream.Client.Chunk.MagicVersion.get -> byte
 RabbitMQ.Stream.Client.Client.QueryRoute(string superStream, string routingKey) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.RouteQueryResponse>
 RabbitMQ.Stream.Client.Client.StreamStats(string stream) -> System.Threading.Tasks.ValueTask<RabbitMQ.Stream.Client.StreamStatsResponse>
+RabbitMQ.Stream.Client.ClientParameters.AuthMechanism.get -> RabbitMQ.Stream.Client.AuthMechanism
+RabbitMQ.Stream.Client.ClientParameters.AuthMechanism.set -> void
 RabbitMQ.Stream.Client.HashRoutingMurmurStrategy.Route(RabbitMQ.Stream.Client.Message message, System.Collections.Generic.List<string> partitions) -> System.Threading.Tasks.Task<System.Collections.Generic.List<string>>
 RabbitMQ.Stream.Client.IConsumerConfig.InitialCredits.get -> ushort
 RabbitMQ.Stream.Client.IConsumerConfig.InitialCredits.set -> void
@@ -53,6 +60,8 @@ RabbitMQ.Stream.Client.StreamStatsResponse.StreamStatsResponse() -> void
 RabbitMQ.Stream.Client.StreamStatsResponse.StreamStatsResponse(uint correlationId, RabbitMQ.Stream.Client.ResponseCode responseCode, System.Collections.Generic.IDictionary<string, long> statistic) -> void
 RabbitMQ.Stream.Client.StreamStatsResponse.Write(System.Span<byte> span) -> int
 RabbitMQ.Stream.Client.StreamSystem.StreamStats(string stream) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.StreamStats>
+RabbitMQ.Stream.Client.StreamSystemConfig.AuthMechanism.get -> RabbitMQ.Stream.Client.AuthMechanism
+RabbitMQ.Stream.Client.StreamSystemConfig.AuthMechanism.set -> void
 static RabbitMQ.Stream.Client.Connection.Create(System.Net.EndPoint endpoint, System.Func<System.Memory<byte>, System.Threading.Tasks.Task> commandCallback, System.Func<string, System.Threading.Tasks.Task> closedCallBack, RabbitMQ.Stream.Client.SslOption sslOption, Microsoft.Extensions.Logging.ILogger logger) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Connection>
 static RabbitMQ.Stream.Client.Message.From(ref System.Buffers.ReadOnlySequence<byte> seq, uint len) -> RabbitMQ.Stream.Client.Message
 static RabbitMQ.Stream.Client.Reliable.DeduplicatingProducer.Create(RabbitMQ.Stream.Client.Reliable.DeduplicatingProducerConfig producerConfig, Microsoft.Extensions.Logging.ILogger<RabbitMQ.Stream.Client.Reliable.Producer> logger = null) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Reliable.DeduplicatingProducer>

RabbitMQ.Stream.Client/StreamSystem.cs

@@ -28,6 +28,8 @@ public record StreamSystemConfig : INamedEntity
 
         public AddressResolver AddressResolver { get; set; }
         public string ClientProvidedName { get; set; } = "dotnet-stream-locator";
+
+        public AuthMechanism AuthMechanism { get; set; } = AuthMechanism.Plain;
     }
 
     public class StreamSystem
@@ -56,7 +58,8 @@ public static async Task<StreamSystem> Create(StreamSystemConfig config, ILogger
                 AddressResolver = config.AddressResolver,
                 ClientProvidedName = config.ClientProvidedName,
                 Heartbeat = config.Heartbeat,
-                Endpoints = config.Endpoints
+                Endpoints = config.Endpoints,
+                AuthMechanism = config.AuthMechanism
             };
             // create the metadata client connection
             foreach (var endPoint in clientParams.Endpoints)
@@ -73,14 +76,19 @@ public static async Task<StreamSystem> Create(StreamSystemConfig config, ILogger
                 }
                 catch (Exception e)
                 {
-                    if (e is ProtocolException or SslException)
+                    switch (e)
                     {
-                        logger?.LogError(e, "ProtocolException or SslException to {@EndPoint}", endPoint);
-                        throw;
+                        case ProtocolException or SslException:
+                            logger?.LogError(e, "ProtocolException or SslException to {@EndPoint}", endPoint);
+                            throw;
+                        case AuthMechanismNotSupportedException:
+                            logger?.LogError(e, "SalsNotSupportedException to {@EndPoint}", endPoint);
+                            throw;
+                        default:
+                            // hopefully all implementations of endpoint have a nice ToString()
+                            logger?.LogError(e, "Error connecting to {@TargetEndpoint}. Trying next endpoint", endPoint);
+                            break;
                     }
-
-                    // hopefully all implementations of endpoint have a nice ToString()
-                    logger?.LogError(e, "Error connecting to {@TargetEndpoint}. Trying next endpoint", endPoint);
                 }
             }
 

Tests/SystemTests.cs

@@ -245,6 +245,17 @@ await Assert.ThrowsAsync<ArgumentException>(
             await system.Close();
         }
 
+        [Fact]
+        public async void ValidateSalsExternalConfiguration()
+        {
+            // the user can set the SALs configuration externally
+            // this test validates that the configuration is supported by the server
+            var config = new StreamSystemConfig() { AuthMechanism = AuthMechanism.External };
+            await Assert.ThrowsAsync<AuthMechanismNotSupportedException>(
+                async () => { await StreamSystem.Create(config); }
+            );
+        }
+
         [Fact]
         public async void CloseProducerConsumerAfterForceCloseShouldNotRaiseError()
         {