Add TLS support (#27)

* Add TLS support

* Use .NET members instead of Java-ish methods.
* Add `UseSsl` to `ConnectionSettings`
*

* * Make `EnsureConnection` async, and use `CreateAsync` to open the actual connection.

* Add a test to try and connect via TLS that fails.

* * Add `ITlsSettings` interface

* Add `TlsSettings` class
* Pass TLS settings to Amqp ConnectionFactory
* Ensure test passes

Author: lukebakken

.gitignore

@@ -14,7 +14,6 @@ nunit-agent*
 test-output.log
 TestResults.xml
 TestResult.xml
-Tests/coverage.xml
 test.sh
 *.VisualState.xml
 .vscode
@@ -121,7 +120,7 @@ projects/Unit*/TestResult.xml
 .*.sw?
 
 # tests
-Tests/coverage.*
+coverage*
 
 # docs
 docs/temp/

RabbitMQ.AMQP.Client/IConnection.cs

@@ -2,7 +2,16 @@
 
 namespace RabbitMQ.AMQP.Client;
 
-public class ConnectionException(string? message) : Exception(message);
+public class ConnectionException : Exception
+{
+    public ConnectionException(string message) : base(message)
+    {
+    }
+
+    public ConnectionException(string message, Exception innerException) : base(message, innerException)
+    {
+    }
+}
 
 public interface IConnection : ILifeCycle
 {
@@ -12,6 +21,5 @@ public interface IConnection : ILifeCycle
 
     IConsumerBuilder ConsumerBuilder();
 
-
     public ReadOnlyCollection<IPublisher> GetPublishers();
 }

RabbitMQ.AMQP.Client/IConnectionSettings.cs

@@ -1,19 +1,50 @@
+using System.Net.Security;
+using System.Security.Authentication;
+using System.Security.Cryptography.X509Certificates;
+
 namespace RabbitMQ.AMQP.Client;
 
-public interface IConnectionSettings
+public interface IConnectionSettings : IEquatable<IConnectionSettings>
 {
-    string Host();
-
-    int Port();
-
-    string VirtualHost();
-
-
-    string User();
-
-    string Password();
-
-    string Scheme();
+    string Host { get; }
+    int Port { get; }
+    string VirtualHost { get; }
+    string User { get; }
+    string Password { get; }
+    string Scheme { get; }
+    string ConnectionName { get; }
+    string Path { get; }
+    bool UseSsl { get; }
+    ITlsSettings? TlsSettings { get; }
+}
 
-    string ConnectionName();
+/// <summary>
+/// Contains the TLS/SSL settings for a connection.
+/// </summary>
+public interface ITlsSettings
+{
+    /// <summary>
+    /// Client certificates to use for mutual authentication.
+    /// </summary>
+    X509CertificateCollection ClientCertificates { get; }
+
+    /// <summary>
+    /// Supported protocols to use.
+    /// </summary>
+    SslProtocols Protocols { get; }
+
+    /// <summary>
+    /// Specifies whether certificate revocation should be performed during handshake.
+    /// </summary>
+    bool CheckCertificateRevocation { get; }
+
+    /// <summary>
+    /// Gets or sets a certificate validation callback to validate remote certificate.
+    /// </summary>
+    RemoteCertificateValidationCallback? RemoteCertificateValidationCallback { get; }
+
+    /// <summary>
+    /// Gets or sets a local certificate selection callback to select the certificate which should be used for authentication.
+    /// </summary>
+    LocalCertificateSelectionCallback? LocalCertificateSelectionCallback { get; }
 }

RabbitMQ.AMQP.Client/Impl/AbstractLifeCycle.cs

@@ -33,7 +33,7 @@ protected void OnNewStatus(State newState, Error? error)
             return;
         }
 
-        var oldStatus = State;
+        State oldStatus = State;
         State = newState;
         ChangeState?.Invoke(this, oldStatus, newState, error);
     }

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -39,7 +39,6 @@ public class AmqpConnection : AbstractLifeCycle, IConnection
     private const string ConnectionNotRecoveredMessage = "Connection not recovered";
     private readonly SemaphoreSlim _semaphoreClose = new(1, 1);
 
-
     // The native AMQP.Net Lite connection
     private Connection? _nativeConnection;
 
@@ -71,7 +70,6 @@ private void ChangeConsumersStatus(State state, Error? error)
         }
     }
 
-
     private async Task ReconnectEntities()
     {
         await ReconnectPublishers().ConfigureAwait(false);
@@ -102,7 +100,6 @@ private async Task ReconnectConsumers()
     // TODO: Implement the semaphore to avoid multiple connections
     // private readonly SemaphoreSlim _semaphore = new(1, 1);
 
-
     /// <summary>
     /// Publishers contains all the publishers created by the connection.
     /// Each connection can have multiple publishers.
@@ -113,7 +110,6 @@ private async Task ReconnectConsumers()
 
     internal ConcurrentDictionary<string, IConsumer> Consumers { get; } = new();
 
-
     public ReadOnlyCollection<IPublisher> GetPublishers()
     {
         return Publishers.Values.ToList().AsReadOnly();
@@ -179,14 +175,18 @@ public IConsumerBuilder ConsumerBuilder()
         return new AmqpConsumerBuilder(this);
     }
 
-    protected override Task OpenAsync()
+    protected override async Task OpenAsync()
     {
-        EnsureConnection();
-        return base.OpenAsync();
+        await EnsureConnection()
+            .ConfigureAwait(false);
+        await base.OpenAsync()
+            .ConfigureAwait(false);
     }
 
-    private void EnsureConnection()
+    private async Task EnsureConnection()
     {
+        // TODO: do this!
+        // await _semaphore.WaitAsync();
         try
         {
             if (_nativeConnection is { IsClosed: false })
@@ -196,22 +196,53 @@ private void EnsureConnection()
 
             var open = new Open
             {
-                HostName = $"vhost:{_connectionSettings.VirtualHost()}",
+                HostName = $"vhost:{_connectionSettings.VirtualHost}",
                 Properties = new Fields()
                 {
-                    [new Symbol("connection_name")] = _connectionSettings.ConnectionName(),
+                    [new Symbol("connection_name")] = _connectionSettings.ConnectionName,
                 }
             };
 
-            var manualReset = new ManualResetEvent(false);
-            _nativeConnection = new Connection(_connectionSettings.Address, null, open, (connection, open1) =>
+            void onOpened(Amqp.IConnection connection, Open open1)
             {
-                manualReset.Set();
                 Trace.WriteLine(TraceLevel.Verbose, $"Connection opened. Info: {ToString()}");
                 OnNewStatus(State.Open, null);
-            });
+            }
+
+            var cf = new ConnectionFactory();
+
+            if (_connectionSettings.UseSsl && _connectionSettings.TlsSettings is not null)
+            {
+                cf.SSL.Protocols = _connectionSettings.TlsSettings.Protocols;
+                cf.SSL.CheckCertificateRevocation = _connectionSettings.TlsSettings.CheckCertificateRevocation;
+
+                if (_connectionSettings.TlsSettings.ClientCertificates.Count > 0)
+                {
+                    cf.SSL.ClientCertificates = _connectionSettings.TlsSettings.ClientCertificates;
+                }
+
+                if (_connectionSettings.TlsSettings.LocalCertificateSelectionCallback is not null)
+                {
+                    cf.SSL.LocalCertificateSelectionCallback = _connectionSettings.TlsSettings.LocalCertificateSelectionCallback;
+                }
+
+                if (_connectionSettings.TlsSettings.RemoteCertificateValidationCallback is not null)
+                {
+                    cf.SSL.RemoteCertificateValidationCallback = _connectionSettings.TlsSettings.RemoteCertificateValidationCallback;
+                }
+            }
+
+            try
+            {
+                _nativeConnection = await cf.CreateAsync(_connectionSettings.Address, open: open, onOpened: onOpened)
+                    .ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                throw new ConnectionException(
+                    $"Connection failed. Info: {ToString()}", ex);
+            }
 
-            manualReset.WaitOne(TimeSpan.FromSeconds(5));
             if (_nativeConnection.IsClosed)
             {
                 throw new ConnectionException(
@@ -294,7 +325,8 @@ await Task.Run(async () =>
                                 await Task.Delay(TimeSpan.FromMilliseconds(next))
                                     .ConfigureAwait(false);
 
-                                EnsureConnection();
+                                await EnsureConnection()
+                                    .ConfigureAwait(false);
                                 connected = true;
                             }
                             catch (Exception e)