Add support for native-tls when using websocket transport

[ondrowan]

This PR adds support for native-tls when using Transport::Wss.

This is still a WIP since the feature tokio-native-tls needs to be set in async-tungstenite when using use-native-tls feature in this crate. At the same time, if use-rustls is set in this crate, async-tungstenite needs to set feature tokio-rustls-native-certs. I have no idea how to edit Cargo.toml to support both of these cases.

At the moment async-tungstenite always uses - even if using insecure WS connection - tokio-rustls-native-certs feature, which I don't think is correct either.

Any help with this or other things I've missed in this PR would be appreciated.

Type of change

New feature (non-breaking change which adds functionality)

Checklist:

• Formatted with cargo fmt
• Make an entry to CHANGELOG.md if it's relevant to the users of the library. If it's not relevant mention why.

[swanandx]

Thank you for the PR!

At the moment async-tungstenite always uses - even if using insecure WS connection - tokio-rustls-native-certs

This is because we specify it here:

async-tungstenite = { version = "0.23", default-features = false, features = ["tokio-rustls-native-certs"], optional = true }

It would be easy to remove this, this feature is required for async_tungstenite::tokio::client_async_tls_with_connector(_) but we can get away without this fn and use async_tungstenite::tokio::client_async_with_config(_) instead ( ofc after doing other required changes ).

Will debug this further, and if it is fine, may I push the changes to tackle this in this PR?

PS: can you provide me minimal example which can be used for testing, thanks :)

[swanandx]

As mentioned in above, I did following changes in eventloop part:

-            let connector = tls::rustls_connector(&tls_config).await?;
+            let stream = tls::tls_connect(&options.broker_addr, options.port, &tls_config, tcp_stream).await?;

-            let (socket, response) = async_tungstenite::tokio::client_async_tls_with_connector(
-                request,
-                tcp_stream,
-                Some(connector),
-            )
-            .await?;
+            let (socket, response) =
+                async_tungstenite::tokio::client_async_with_config(request, stream, None).await?;

And then to try it out, I was using code similar to this:

#[tokio::main(worker_threads = 1)]
async fn main() -> Result<(), Box<dyn Error>> {
    pretty_env_logger::init();

    // port parameter is ignored when scheme is websocket
    let mut mqttoptions = MqttOptions::new("client", "wss://localhost:8083", 8083);

    let ca = include_str!("../../../ca.cert.pem");
    let client_cert = include_str!("../../../device1.cert.pem");
    let client_key = include_str!("../../../device1.key.pem");
    let transport = Transport::Wss(TlsConfiguration::Simple {
        ca: ca.into(),
        alpn: None,
        client_auth: Some((client_cert.into(), Key::RSA(client_key.into()))),
    });

    mqttoptions.set_transport(transport);
    mqttoptions.set_keep_alive(Duration::from_secs(60));

    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
    // rest of the code
}

but I am getting Error = Tls(DNSName(InvalidDnsNameError)) . any idea what might be causing this invalid dns name error? ( note: certs i am using works fine with normal tcp + tls connection, so it must be something with async_tungstenite ig )

[ondrowan]

I have tried to implement the changes you've suggested and ended up getting Tls(NativeTls(Error { code: -67843, message: "The certificate was not trusted." })) for some reason. I've used almost the same example as yours with the exception of:

mqttoptions.set_transport(Transport::wss_with_config(
    rumqttc::TlsConfiguration::Native,
));

As for the example, I'm trying the code to connect to Azure IoT hub which requires TLS 1.0 (at least in EU region) and the complete code example looks like this:

use rumqttc::{AsyncClient, MqttOptions, QoS, Transport};
use std::error::Error;
use std::time::Duration;
use tokio::task;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    env_logger::init();

    let device_id = "...";
    let url = "...";
    let publish_queue = format!("devices/{device_id}/messages/events/");

    let mut mqttoptions = MqttOptions::new(
        "...",
        format!("wss://{url}:443/$iothub/websocket"),
        443,
    );
    mqttoptions.set_credentials(
        format!("{url}/{device_id}/?api-version=2021-04-12"),
        "..."
    );

    mqttoptions.set_transport(Transport::wss_with_config(
        rumqttc::TlsConfiguration::Native,
    ));
    mqttoptions.set_keep_alive(Duration::from_secs(120));

    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);

    task::spawn(async move {
        let msg = r#"..."#;

        for _ in 0..10 {
            client
                .publish(&publish_queue, QoS::AtLeastOnce, false, msg)
                .await
        }
    });

    loop {
        let event = eventloop.poll().await;
        match event {
            Ok(notification) => println!("Received = {:?}", notification),
            Err(error) => println!("Error = {:?}", error),
        }
    }
}

[swanandx]

Have a look over this issue and reddit post. This might be the case here!

[ondrowan]

I'm afraid all this stuff is completely outside of my expertise and I won't be able to finish this PR. I wanted to give it a try, but I don't even know how to generate proper certificates to use with rumqttc::TlsConfiguration::SimpleNative.

[swanandx]

Hey @ondrowan , is it fine if I push some changes here? just want to sync up with main.

[ondrowan]

@swanandx As I've said I'm pretty lost in this certificate stuff, so go ahead.

As a sidenote, the code I've submitted works fine when I'm using it with

mqttoptions.set_transport(Transport::wss_with_config(
    rumqttc::TlsConfiguration::Native,
));

At least on Linux.

[swanandx]

so with the code I pushed, and with emqx public broker:

use rumqttc::{self, AsyncClient, MqttOptions, QoS, TlsConfiguration, Transport};
use std::{error::Error, time::Duration};

#[tokio::main(worker_threads = 1)]
async fn main() -> Result<(), Box<dyn Error>> {
    pretty_env_logger::init();

    // port parameter is ignored when scheme is websocket
    let mut mqttoptions =
        MqttOptions::new("clientId-aSziq39Bp3", "wss://broker.emqx.io:8084/", 8084);

    // using rustls
    // Error = Websocket(Io(Custom { kind: InvalidData, error: InvalidCertificate(UnknownIssuer) }))
    // let tls = TlsConfiguration::Simple {
    //     ca: include_bytes!("/home/swanx/bytebeam/certs/broker.emqx.io-ca.crt").into(),
    //     alpn: None,
    //     client_auth: None,
    // };
    
    // using native-tls
    // Error = Websocket(Http(Response { status: 404, version: HTTP/1.1, headers: {"content-length": "0", "date": "Thu, 07 Dec 2023 17:41:33 GMT", "server": "Cowboy"}, body: Some([]) }))
    let tls = TlsConfiguration::SimpleNative {
        ca: include_bytes!("/home/swanx/bytebeam/certs/broker.emqx.io-ca.crt").into(),
        client_auth: None,
    };
    let transport = Transport::Wss(tls);
    mqttoptions.set_transport(transport);
    mqttoptions.set_keep_alive(Duration::from_secs(60));

    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
    client
        .publish("hello/world", QoS::ExactlyOnce, false, "test")
        .await
        .unwrap();

    loop {
        let event = eventloop.poll().await;
        match event {
            Ok(notif) => {
                println!("Event = {notif:?}");
            }
            Err(err) => {
                println!("Error = {err:?}");
                return Ok(());
            }
        }
    }
}

note: on TLS port 8884 ( without websocket ) I am getting same error for rustls, but for native-tls it works fine!

I did try :

    let transport = Transport::wss_with_config(rumqttc::TlsConfiguration::Native);

to connect with emqx, and it gave same 404 error.
With hivemq ( with TlsConfiguration::Native ) it get Error = NetworkTimeout over Wss ( works fine with Tls )

[swanandx]

I will try once again with generating certs on my own tomorrow!

[swanandx]

I used provision to generate certs:

# generate ca certs
$ ./provision ca
# generate server certs for rumqttd
$ ./provision server --ca ca.cert.pem --cakey ca.key.pem --domain "localhost"

if you with to use native-tls in broker side as well ( note: nothing to do with rumqttc end! ), you can use openssl to get pkcs12 like:

$ openssl pkcs12 -export -out pkcs12_localhost.pfx -in localhost.cert.pem -inkey localhost.key.pem -certfile ca.cert.pem

Then in rumqttd config: ( note: update path to your certs )

[ws.2]
name = "ws-2"
listen = "0.0.0.0:8081"
next_connection_delay_ms = 1
    [ws.2.tls]
    # if using rumqttd with rustls
    capath = "/home/swanx/bytebeam/certs/ca.cert.pem"
    certpath = "/home/swanx/bytebeam/certs/localhost.cert.pem"
    keypath = "/home/swanx/bytebeam/certs/localhost.key.pem"
    # if using rumqttd with native tls!
    # pkcs12path = "/home/swanx/bytebeam/certs/pkcs12_localhost.pfx"
    # pkcs12pass = "" # enter pass you entered while generating cert using openssl
    [ws.2.connections]
    # rest of the options
    # ...

with this setup in rumqttd, I was able to connect with it successfully with both native-tls and rustls!

//...
let mut mqttoptions = MqttOptions::new("clientId-aSziq39Bp3", "wss://localhost:8081/", 8081);

// using rustls
// let tls = TlsConfiguration::Simple {
//     ca: include_bytes!("/home/swanx/bytebeam/certs/ca.cert.pem").into(),
//     alpn: None,
//     client_auth: None,
// };

// using native-tls
let tls = TlsConfiguration::SimpleNative {
     ca: include_bytes!("/home/swanx/bytebeam/certs/ca.cert.pem").into(),
     client_auth: None,
 };

let transport = Transport::Wss(tls);
// ...

so basically it is working! can you verify it latest changes in the PR solves your original issue?

[ondrowan]

I've just tested it with Azure IoT Hub with the following setup:

mqttoptions.set_transport(Transport::wss_with_config(
    rumqttc::TlsConfiguration::Native,
));

and it works!

[swanandx]

That's great to hear!

Need to figure out a way to tackle issue caused due to enabling both features together! ref: sdroege/async-tungstenite#78 (comment)