SASL mechanism and RabbitCrDemo example

RabbitMQ.Next.sln

@@ -50,6 +50,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "RabbitMQ.Next.Serialization
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "RabbitMQ.Next.Examples.DynamicSerializer", "docs\examples\RabbitMQ.Next.Examples.DynamicSerializer\RabbitMQ.Next.Examples.DynamicSerializer.csproj", "{B93A51A9-0A31-4419-B8E8-C304EF18CFAB}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "RabbitMQ.Next.Examples.DemoSaslAuthMechanism", "docs\examples\RabbitMQ.Next.Examples.DemoSaslAuthMechanism\RabbitMQ.Next.Examples.DemoSaslAuthMechanism.csproj", "{C0440748-9787-4891-874A-054F5E64020B}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -136,6 +138,10 @@ Global
 		{B93A51A9-0A31-4419-B8E8-C304EF18CFAB}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{B93A51A9-0A31-4419-B8E8-C304EF18CFAB}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{B93A51A9-0A31-4419-B8E8-C304EF18CFAB}.Release|Any CPU.Build.0 = Release|Any CPU
+		{C0440748-9787-4891-874A-054F5E64020B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{C0440748-9787-4891-874A-054F5E64020B}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{C0440748-9787-4891-874A-054F5E64020B}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{C0440748-9787-4891-874A-054F5E64020B}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(NestedProjects) = preSolution
 		{AB00A6F3-C3E0-4E61-B302-F9CF0288819E} = {C895EE24-13DF-45C3-AF1D-B6B6EE4A10A9}
@@ -160,5 +166,6 @@ Global
 		{6D34D919-0630-40EB-9291-CB28DB2BBA25} = {AD3AF1AA-71EE-4A2C-84DD-5C7DF141C187}
 		{95003EED-4B62-430E-8A4D-1D3B09CAA173} = {AD3AF1AA-71EE-4A2C-84DD-5C7DF141C187}
 		{B93A51A9-0A31-4419-B8E8-C304EF18CFAB} = {750B1D71-E43B-4714-8AA3-AB9B0EC7E931}
+		{C0440748-9787-4891-874A-054F5E64020B} = {750B1D71-E43B-4714-8AA3-AB9B0EC7E931}
 	EndGlobalSection
 EndGlobal

docs/examples/RabbitMQ.Next.Examples.DemoSaslAuthMechanism/ConnectionBuilderExtensions.cs

@@ -0,0 +1,10 @@
+namespace RabbitMQ.Next.Examples.DemoSaslAuthMechanism;
+
+public static class ConnectionBuilderExtensions
+{
+    public static IConnectionBuilder WithRabbitCrDemoAuth(this IConnectionBuilder builder, string userName, string password)
+    {
+        builder.Auth(new RabbitCrDemoAuthMechanism(userName, password));
+        return builder;
+    }
+}

docs/examples/RabbitMQ.Next.Examples.DemoSaslAuthMechanism/Program.cs

@@ -0,0 +1,23 @@
+using System;
+using System.Threading.Tasks;
+
+namespace RabbitMQ.Next.Examples.DemoSaslAuthMechanism;
+
+class Program
+{
+    static async Task Main()
+    {
+        Console.WriteLine("Hello World! Will try to connect RabbitMQ server with RABBIT-CR-DEMO auth mechanism.");
+
+        var connection = await ConnectionBuilder.Default
+            .Endpoint("amqp://localhost:5672/")
+            .WithRabbitCrDemoAuth("guest", "guest")
+            .ConnectAsync()
+            .ConfigureAwait(false);
+
+        Console.WriteLine("Connection opened");
+        Console.WriteLine("Press any key to close the connection");
+
+        Console.ReadKey();
+    }
+}

docs/examples/RabbitMQ.Next.Examples.DemoSaslAuthMechanism/RabbitCrDemoAuthMechanism.cs

@@ -0,0 +1,35 @@
+using System;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace RabbitMQ.Next.Examples.DemoSaslAuthMechanism;
+
+internal class RabbitCrDemoAuthMechanism : IAuthMechanism
+{
+    private readonly string username;
+    private readonly string password;
+
+    public RabbitCrDemoAuthMechanism(string username, string password)
+    {
+        this.username = username;
+        this.password = password;
+    }
+
+    public string Type => "RABBIT-CR-DEMO";
+
+    public ValueTask<ReadOnlyMemory<byte>> StartAsync() 
+        => ValueTask.FromResult(new ReadOnlyMemory<byte>(Encoding.UTF8.GetBytes(this.username)));
+
+
+    public ValueTask<ReadOnlyMemory<byte>> HandleChallengeAsync(ReadOnlySpan<byte> challenge)
+    {
+        var serverChallenge = Encoding.UTF8.GetString(challenge);
+
+        if (string.Equals("Please tell me your password", serverChallenge))
+        {
+            return ValueTask.FromResult(new ReadOnlyMemory<byte>(Encoding.UTF8.GetBytes($"My password is {this.password}")));    
+        }
+
+        throw new InvalidOperationException();
+    }
+}

docs/examples/RabbitMQ.Next.Examples.DemoSaslAuthMechanism/RabbitMQ.Next.Examples.DemoSaslAuthMechanism.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net6.0</TargetFramework>
+    <IsPackable>false</IsPackable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\src\RabbitMQ.Next.Abstractions\RabbitMQ.Next.Abstractions.csproj" />
+    <ProjectReference Include="..\..\..\src\RabbitMQ.Next\RabbitMQ.Next.csproj" />
+  </ItemGroup>
+
+</Project>

src/RabbitMQ.Next.Abstractions/Auth/PlainAuthMechanism.cs

@@ -4,7 +4,7 @@
 
 namespace RabbitMQ.Next.Auth;
 
-public class PlainAuthMechanism : IAuthMechanism
+internal class PlainAuthMechanism : IAuthMechanism
 {
     public PlainAuthMechanism(string userName, string password)
     {
@@ -13,16 +13,15 @@ public PlainAuthMechanism(string userName, string password)
     }
 
     public string Type => "PLAIN";
-    public ValueTask<ReadOnlyMemory<byte>> HandleChallengeAsync(ReadOnlySpan<byte> challenge)
-    {
-        if (!challenge.IsEmpty)
-        {
-            throw new NotSupportedException("PlainAuthMechanism does not support challenges.");
-        }
 
+    public ValueTask<ReadOnlyMemory<byte>> StartAsync()
+    {
         ReadOnlyMemory<byte> response = Encoding.UTF8.GetBytes($"\0{this.UserName}\0{this.Password}");
         return ValueTask.FromResult(response);
     }
+
+    public ValueTask<ReadOnlyMemory<byte>> HandleChallengeAsync(ReadOnlySpan<byte> challenge) 
+        => throw new NotSupportedException("PlainAuthMechanism does not support challenges.");
 
     public string UserName { get; }
 

src/RabbitMQ.Next.Abstractions/IAuthMechanism.cs

@@ -6,6 +6,8 @@ namespace RabbitMQ.Next;
 public interface IAuthMechanism
 {
     string Type { get; }
-
+
+    ValueTask<ReadOnlyMemory<byte>> StartAsync();
+
     ValueTask<ReadOnlyMemory<byte>> HandleChallengeAsync(ReadOnlySpan<byte> challenge);
 }

src/RabbitMQ.Next.Publisher.Abstractions/PublisherExtensions.cs

@@ -13,11 +13,12 @@ public static Task PublishAsync<TContent>(
         Action<IMessageBuilder> propertiesBuilder = null,
         CancellationToken cancellation = default)
         => publisher.PublishAsync(
-            routingKey,
+            (routingKey,propertiesBuilder),
             content,
             (state, message) =>
             {
-                message.SetRoutingKey(state);
+                message.SetRoutingKey(state.routingKey);
+                state.propertiesBuilder?.Invoke(message);
             },
             cancellation);
 }

src/RabbitMQ.Next/Connection.cs

@@ -276,10 +276,28 @@ private static async Task<NegotiationResults> NegotiateConnectionAsync(IChannel
             throw new NotSupportedException("Provided auth mechanism does not supported by the server");
         }
 
-        var initialChallengeResponse = await settings.Auth.HandleChallengeAsync(Span<byte>.Empty);
+        var saslStartBytes = await settings.Auth.StartAsync();
 
         var tuneMethodTask = channel.WaitAsync<TuneMethod>(cancellation);
-        await channel.SendAsync(new StartOkMethod(settings.Auth.Type, initialChallengeResponse, settings.Locale, settings.ClientProperties), cancellation);
+        var secureMethodTask = channel.WaitAsync<SecureMethod>(cancellation);
+
+        await channel.SendAsync(new StartOkMethod(settings.Auth.Type, saslStartBytes, settings.Locale, settings.ClientProperties), cancellation);
+
+        do
+        {
+            await Task.WhenAny(tuneMethodTask, secureMethodTask);
+
+            if (secureMethodTask.IsCompleted)
+            {
+                var secureRequest = await secureMethodTask;
+                var secureResponse = await settings.Auth.HandleChallengeAsync(secureRequest.Challenge.Span);
+
+                // wait for another secure round-trip just in case
+                secureMethodTask = channel.WaitAsync<SecureMethod>(cancellation);
+                await channel.SendAsync(new SecureOkMethod(secureResponse), cancellation);
+            }
+
+        } while (!tuneMethodTask.IsCompleted);
 
         var tuneMethod = await tuneMethodTask;
         var negotiationResult = new NegotiationResults(

tests/RabbitMQ.Next.Tests/Abstractions/Auth/PlainAuthMechanismTests.cs

@@ -23,17 +23,25 @@ public void CtorTests()
     }
 
     [Fact]
-    public async Task HandleChallengeAsync()
+    public async Task StartAsyncTests()
     {
         var user = "test";
         var password = "pwd";
         var expected = "\0test\0pwd"u8.ToArray();
 
         var auth = new PlainAuthMechanism(user, password);
-        var response = await auth.HandleChallengeAsync(ReadOnlySpan<byte>.Empty);
+        var response = await auth.StartAsync();
 
         Assert.Equal(expected, response.ToArray());
     }
+
+    [Fact]
+    public async Task HandleChallengeAsyncThrows()
+    {
+        var auth = new PlainAuthMechanism("test", "pwd");
+
+        await Assert.ThrowsAsync<NotSupportedException>(async () => await auth.HandleChallengeAsync(ReadOnlySpan<byte>.Empty));
+    }
 
     [Fact]
     public void ExtensionTests()

tests/RabbitMQ.Next.Tests/Abstractions/ConnectionBuilderExtensionsTests.cs

@@ -46,51 +46,51 @@ public void AddEndpointUriThrowsOnNoAmqp()
 
     [Theory]
     [MemberData(nameof(AddEndpointTestCases))]
-    public void AddEndpointCanParseValidUri(string endpoint, bool ssl, string host, int port, string vhost, PlainAuthMechanism auth)
+    public void AddEndpointCanParseValidUri(string endpoint, bool ssl, string host, int port, string vhost, string userName, string password)
     {
         var builder = Substitute.For<IConnectionBuilder>();
 
         builder.Endpoint(endpoint);
 
         builder.Received().Endpoint(host, port, ssl);
         builder.Received().VirtualHost(vhost);
-        if (auth == null)
+        if (string.IsNullOrEmpty(userName))
         {
             builder.DidNotReceive().Auth(Arg.Any<IAuthMechanism>());
         }
         else
         {
-            builder.Received().Auth(Arg.Is<PlainAuthMechanism>(x => x.Type == "PLAIN" && x.UserName == auth.UserName && x.Password == auth.Password));
+            builder.Received().Auth(Arg.Is<PlainAuthMechanism>(x => x.Type == "PLAIN" && x.UserName == userName && x.Password == password));
         }
     }
 
     public static IEnumerable<object[]> AddEndpointTestCases()
     {
         yield return new object[] {"amqp://user:pass@host:10000/vhost",
-            false, "host", 10000, "vhost", new PlainAuthMechanism("user", "pass")};
+            false, "host", 10000, "vhost", "user", "pass"};
 
         yield return new object[] {"AMQP://user:pass@host:10000/vhost",
-            false, "host", 10000, "vhost", new PlainAuthMechanism("user", "pass")};
+            false, "host", 10000, "vhost", "user", "pass"};
 
         yield return new object[] {"amqp://user%61:%61pass@ho%61st:10000/v%2fhost",
-            false, "hoast", 10000, "v/host", new PlainAuthMechanism("usera", "apass")};
+            false, "hoast", 10000, "v/host", "usera", "apass"};
 
         yield return new object[] {"amqp://user@localhost",
-            false, "localhost", 5672, "/", new PlainAuthMechanism("user", "") };
+            false, "localhost", 5672, "/", "user", "" };
 
         yield return new object[] {"amqp://[::1]",
-            false, "[::1]", 5672, "/", null};
+            false, "[::1]", 5672, "/", null, null};
 
         yield return new object[] {"amqps://user:pass@host:10000/vhost",
-            true, "host", 10000, "vhost", new PlainAuthMechanism("user", "pass")};
+            true, "host", 10000, "vhost", "user", "pass"};
 
         yield return new object[] {"amqps://user%61:%61pass@ho%61st:10000/v%2fhost",
-            true, "hoast", 10000, "v/host", new PlainAuthMechanism("usera", "apass")};
+            true, "hoast", 10000, "v/host", "usera", "apass"};
 
         yield return new object[] {"amqps://user@localhost",
-            true, "localhost", 5671, "/", new PlainAuthMechanism("user", "")};
+            true, "localhost", 5671, "/", "user", ""};
 
         yield return new object[] {"amqps://[::1]",
-            true, "[::1]", 5671, "/", null};
+            true, "[::1]", 5671, "/", null, null};
     }
 }