Exception during AWS connection

[gabrielvictorafreitas]

Hello everyone!

I'm trying to connect into an AWS Broker using .net6 and MqttNet 4.3.6.1152 and i'm receiving this exception:

It's good to mention that I'm testing on WSL Ubuntu 20.04 and production server is a Linux Ubuntu 20.04!

Exception has occurred: CLR/MQTTnet.Exceptions.MqttCommunicationException An unhandled exception of type 'MQTTnet.Exceptions.MqttCommunicationException' occurred in System.Private.CoreLib.dll: ' Received an unexpected EOF or 0 bytes from the transport stream.' Inner exceptions found, see $exception in variables window for more details. Innermost exception System.IO.IOException : Received an unexpected EOF or 0 bytes from the transport stream. at System.Net.Security.SslStream.<<FillHandshakeBufferAsync>g__InternalFillHandshakeBufferAsync|189_0>d1.MoveNext()
at System.Runtime.CompilerServices.ConfiguredValueTaskAwaitable1.ConfiguredValueTaskAwaiter.GetResult() at System.Net.Security.SslStream.<ReceiveBlobAsync>d__1761.MoveNext()
at System.Threading.Tasks.ValueTask1.get_Result() at System.Runtime.CompilerServices.ConfiguredValueTaskAwaitable1.ConfiguredValueTaskAwaiter.GetResult()
at System.Net.Security.SslStream.d__1751.MoveNext() at MQTTnet.Implementations.MqttTcpChannel.<ConnectAsync>d__17.MoveNext() at MQTTnet.Implementations.MqttTcpChannel.<ConnectAsync>d__17.MoveNext() at MQTTnet.Adapter.MqttChannelAdapter.<>c__DisplayClass30_0.<<ConnectAsync>b__1>d.MoveNext() at MQTTnet.Adapter.MqttChannelAdapter.<ConnectAsync>d__30.MoveNext()

This is my code:

`private async Task StartMqttServer(CancellationToken cancellationToken)
{

        // Create a new MQTT client.
        var factory = new MqttFactory();
        var mqttClient = factory.CreateMqttClient();


        var certBytes = Encoding.UTF8.GetBytes(caCertificate);
        var signingcert = new X509Certificate2(certBytes);

        var reader = new CertificateFromPemReader();
        X509Certificate2 deviceCertificate = reader.LoadCertificateWithPrivateKeyFromStrings(certificate, privateKey);



        IEnumerable<X509Certificate2> certs = new List<X509Certificate2>
        {
            signingcert,
            deviceCertificate
        };


        var mqttClientOptions = new MqttClientOptionsBuilder()
                        .WithTcpServer(brokerAddress)
                        .WithClientId(Guid.NewGuid().ToString())
                        .WithoutPacketFragmentation()
                        .WithTlsOptions(
                            o =>
                            {
                                o.WithCertificateValidationHandler(_ => true);
                                o.WithClientCertificates(certs);
                                o.WithSslProtocols(SslProtocols.Tls12);
                                o.WithAllowUntrustedCertificates(true);
                            })
                        .Build();



            var response = await mqttClient.ConnectAsync(mqttClientOptions, cancellationToken);
            if (response.ResultCode == MqttClientConnectResultCode.Success)
            {
                _logger.LogInformation("Connected to MQTT broker successfully.");
                Console.WriteLine("Connected to MQTT broker successfully.");


                var mqttSubscribeOptions = factory.CreateSubscribeOptionsBuilder()
                               .WithTopicFilter(
                                   f =>
                                   {
                                       f.WithTopic(topic);
                                   })
                               .Build();

                // Subscribe to a topic
                await mqttClient.SubscribeAsync(mqttSubscribeOptions, cancellationToken).ConfigureAwait(false);


                // Callback function when a message is received
                mqttClient.ApplicationMessageReceivedAsync += e =>
               {
                   async Task ProcessAsync()
                   {
                       string payload = Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment);
                       Console.WriteLine($"Received message console: {payload}");

                       _logger.LogInformation($"Received message: {payload}");
                       Save(payload, out _);
                       await Task.Delay(1000, cancellationToken);
                       await e.AcknowledgeAsync(cancellationToken);
                   }
                   _ = Task.Run(ProcessAsync, cancellationToken);

                   return Task.CompletedTask;
               };
            }
            else
            {
                Console.WriteLine("Not Connected to MQTT broker.");
                _logger.LogInformation("Not Connected to MQTT broker.");
            }
    }

`

Any help please?

[rido-min]

hi @gabrielvictorafreitas

This error happens when you try to connect to an endpoint without TLS enabled, make sure you are using the right brokerAddress

I double checked with the code below and I'm able to connect to IoTCore in AWS

var factory = new MqttFactory();
var mqttClient = factory.CreateMqttClient();
string brokerAddress = "******-ats.iot.us-west-1.amazonaws.com";
X509Certificate2 cert = X509Certificate2.CreateFromPemFile("cert.pem", "private.key");
IEnumerable<X509Certificate2> certs = new List<X509Certificate2>
{
    // see https://github.com/dotnet/runtime/issues/45680#issuecomment-739912495
    new X509Certificate2(cert.Export(X509ContentType.Pkcs12))
};

var mqttClientOptions = new MqttClientOptionsBuilder()
                .WithTcpServer(brokerAddress)
                .WithClientId("basicPubsub")
                .WithTlsOptions(o => o.WithClientCertificates(certs))
                .Build();
var response = await mqttClient.ConnectAsync(mqttClientOptions);
Console.WriteLine(response.ResultCode);

Some comments from your code:

1. You dont need to disable certificate validation (with AllowUntrustedCertificates)
2. .NET in WSL is able to validate the AWS CA from trusted roots, so you dont need to configure it
3. The ClientId has to be configured in the AWS Policy, otherwise you might get a NonAuthorized result
4. When loading the private key, you need to export/import, as described in .NET Core 5.0: X509Certificate2 - No credentials are available in the security package runtime#45680 (comment)

[rido-min]

the topic configuration depends on the broker, not sure about AWS IoTCore.

[gabrielvictorafreitas]

I know that we need to configure something like this:
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder() .WithTopicFilter( f => { f.WithTopic("mqttnet/samples/topic/1"); }) .WithTopicFilter( f => { f.WithTopic("mqttnet/samples/topic/2").WithNoLocal(); }) .WithTopicFilter( f => { f.WithTopic("mqttnet/samples/topic/3").WithRetainHandling(MqttRetainHandling.SendAtSubscribe); }) .Build();

my ideia is to do something like this:

`List topics = new List(topic.Split('|'));

            var mqttSubscribeOptions = factory.CreateSubscribeOptionsBuilder()
                           .WithTopicFilter(
                               f =>
                               {
                                   foreach (var t in topics)
                                   {
                                    
                                   f.WithTopic(t);
                                   }
                               })
                           .Build();`

I know that this configuration is wrong

[rido-min]

do you want to open a new discussion? this one was about authentication to AWS.

[gabrielvictorafreitas]

Sure

[gabrielvictorafreitas]

@rido-min #2015

[gabrielvictorafreitas]

@rido-min Thank you a lot! Now it's working!

May main issue was this part:
IEnumerable<X509Certificate2> certs = new List<X509Certificate2> { // see https://github.com/dotnet/runtime/issues/45680#issuecomment-739912495 new X509Certificate2(cert.Export(X509ContentType.Pkcs12)) };

And I followed your comments!