Support External SASL configuration

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()
         {

docs/Documentation/StreamSystemUsage.cs

@@ -62,6 +62,39 @@ private static async Task CreateTls()
         await streamSystem.Close().ConfigureAwait(false); // <2>
     }
     // end::create-tls[]
+
+
+    // tag::create-tls-external-auth[]
+    private static async Task CreateTlsExternal()
+    {
+        var ssl = new SslOption() // <1>
+        {
+            Enabled = true,
+            ServerName = "server_name",
+            CertPath = "certs/client/keycert.p12",
+            CertPassphrase = null, // in case there is no password
+            CertificateValidationCallback = (sender, certificate, chain, errors) => true,
+        };
+
+        var config = new StreamSystemConfig()
+        {
+            UserName = "user_does_not_exist",
+            Password = "password_does_not_exist",
+            Ssl = ssl,
+            Endpoints = new List<EndPoint>(new List<EndPoint>()
+            {
+                new DnsEndPoint("server_name", 5551)
+            }),
+
+            AuthMechanism = AuthMechanism.External, // <2>
+        };
+
+        var streamSystem = await StreamSystem.Create(config).ConfigureAwait(false);
+
+        await streamSystem.Close().ConfigureAwait(false);
+
+    }
+    // end::create-tls-external-auth[]
 
 
     // tag::create-tls-trust[]

docs/asciidoc/api.adoc

@@ -66,7 +66,19 @@ include::{test-examples}/StreamSystemUsage.cs[tag=create-tls]
 --------
 
 <1> Enable TLS
-<2> Load certificate authority (CA) certificate from PEM file
+<2> Load certificates from PEM files
+
+.Creating an StreamSystem that uses TLS and external authentication
+[source,c#,indent=0]
+--------
+include::{test-examples}/StreamSystemUsage.cs[tag=create-tls-external-auth]
+--------
+
+<1> Enable TLS and configure the certificates
+<2> Set the external authentication mechanism
+
+Note: you need the `rabbitmq_auth_mechanism_ssl` plugin enabled on the server side to use external authentication. 
+`AuthMechanism.External` can be used from RabbitMQ server 3.11.19 and RabbitMQ 3.12.1 onwards.
 
 .Creating a TLS environment that trusts all server certificates for development
 [source,c#,indent=0]
@@ -76,6 +88,9 @@ include::{test-examples}/StreamSystemUsage.cs[tag=create-tls-trust]
 
 <1> Trust all server certificates
 
+
+
+
 ===== Configuring the Stream System
 
 The following table sums up the main settings to create an `StreamSystem` using the `StreamSystemConfig`: