Support for MQTTS (#43)

Cargo.toml

@@ -28,6 +28,7 @@ categories = ["network-programming"]
 async-channel = "2.2.0"
 async-std = "=1.12.0"
 async-trait = "0.1.66"
+base64 = "0.21.4"
 clap = "3.2.23"
 derivative = "2.2.0"
 env_logger = "0.10.0"
@@ -39,8 +40,12 @@ lazy_static = "1.4.0"
 log = "0.4.17"
 ntex = "0.7.17"
 ntex-mqtt = "0.12.16"
+ntex-tls = "0.3.2"
 regex = "1.7.1"
 rustc_version = "0.4"
+rustls = "0.21.7"
+rustls-pemfile = "1.0.4"
+secrecy = { version = "0.8.0", features = ["serde", "alloc"] }
 serde = "1.0.154"
 serde_json = "1.0.94"
 zenoh = { version = "0.11.0-dev", git = "https://github.com/eclipse-zenoh/zenoh.git", branch = "main", features = [

DEFAULT_CONFIG.json5

@@ -45,6 +45,33 @@
       ////
       // generalise_subs: ["PUB1", "PUB2"],
 
+      ////
+      //// TLS related configuration (MQTTS active only if this part is defined).
+      ////
+      // tls: {
+      //   ////
+      //   //// server_private_key: TLS private key provided as either a file or base 64 encoded string.
+      //   ////                     One of the values below must be provided.
+      //   ////
+      //   // server_private_key: "/path/to/private-key.pem",
+      //   // server_private_key_base64: "base64-private-key",
+      //
+      //   ////
+      //   //// server_certificate: TLS public certificate provided as either a file or base 64 encoded string.
+      //   ////                     One of the values below must be provided.
+      //   ////
+      //   // server_certificate: "/path/to/certificate.pem",
+      //   // server_certificate_base64: "base64-certificate",
+      //
+      //   ////
+      //   //// root_ca_certificate: Certificate of the certificate authority used to validate clients connecting to the MQTT server.
+      //   ////                      Provided as either a file or base 64 encoded string.
+      //   ////                      This setting is optional and enables mutual TLS (mTLS) support if provided.
+      //   ////
+      //   // root_ca_certificate: "/path/to/root-ca-certificate.pem",
+      //   // root_ca_certificate_base64: "base64-root-ca-certificate",
+      // },
+
     },
 
     ////

README.md

@@ -72,6 +72,9 @@ The `"mqtt"` part of this same configuration file can also be used in the config
    - **`-r, --generalise-sub <String>`** :  A list of key expressions to use for generalising the declaration of
      the zenoh subscriptions, and thus minimizing the discovery traffic (usable multiple times).
      See [this blog](https://zenoh.io/blog/2021-03-23-discovery/#leveraging-resource-generalisation) for more details.
+   - **`--server-private-key <FILE>`** :  Path to the TLS private key for the MQTT server. If specified a valid certificate for the server must also be provided.
+   - **`--server-certificate <FILE>`** :  Path to the TLS public certificate for the MQTT server. If specified a valid private key for the server must also be provided.
+   - **`--root-ca-certificate <FILE>`** :  Path to the certificate of the certificate authority used to validate clients connecting to the MQTT server. If specified a valid private key and certificate for the server must also be provided.
 
 ## Admin space
 
@@ -89,6 +92,60 @@ Example of queries on administration space using the REST API with the `curl` co
 
 > _Pro tip: pipe the result into [**jq**](https://stedolan.github.io/jq/) command for JSON pretty print or transformation._
 
+## MQTTS support
+
+The MQTT plugin and standalone bridge for Eclipse Zenoh supports MQTTS. MQTTS can be configured in two ways:
+
+ - server side authentication: MQTT clients validate the servers TLS certificate but not the other way around.
+ - mutual authentication (mTLS): where both server and clients validate each other.
+
+ MQTTS can be configured via the configuration file or, if using the standalone bridge, via command line arguments.
+
+### Server side authentication configuration
+
+Server side authentication requires both a private key and certificate for the server. These can be provided as either a file or as a base 64 encoded string.
+
+In the configuration file, the required **tls** fields when using files are **server_private_key** and **server_certificate**. When using base 64 encoded strings the required **tls** fields are **server_private_key_base64** and **server_certificate_base64**.
+
+An example configuration file supporting server side authentication would be:
+
+```json
+{
+  "plugins": {
+    "mqtt": {
+      "tls": {
+        "server_private_key": "/path/to/private-key.pem",
+        "server_certificate": "/path/to/certificate.pem"
+      }
+    }
+  }
+}
+```
+
+The standalone bridge (`zenoh-bridge-mqtt`) also allows the required files to be provided through the **`--server-private-key`** and **`--server-certificate`** command line arguments.
+
+### Mutual authentication (mTLS) configuration
+
+In order to enable mutual authentication a certificate for the certificate authority used to validate clients connecting to the MQTT server must also be provided. This can be provided as either a file or a base 64 encoded string.
+
+In the configuration file, the required **tls** field when using a file is **root_ca_certificate**. When using base 64 encoded strings the required **tls** field when using a file is **root_ca_certificate_base64**.
+
+An example configuration file supporting server side authentication would be:
+
+```json
+{
+  "plugins": {
+    "mqtt": {
+      "tls": {
+        "server_private_key": "/path/to/private-key.pem",
+        "server_certificate": "/path/to/certificate.pem",
+        "root_ca_certificate": "/path/to/root-ca-certificate.pem"
+      }
+    }
+  }
+}
+```
+The standalone bridge (`zenoh-bridge-mqtt`) also allows the required file to be provided through the **`--root-ca-certificate`** command line argument.
 
 ## How to install it
 

zenoh-bridge-mqtt/src/main.rs

@@ -104,7 +104,19 @@ r#"-r, --generalise-sub=[String]...   'A list of key expression to use for gener
         ))
         .arg(Arg::from_usage(
 r#"-w, --generalise-pub=[String]...   'A list of key expression to use for generalising publications (usable multiple times).'"#
-        ));
+        ))
+        .arg(Arg::from_usage(
+r#"--server-private-key=[FILE]   'Path to the TLS private key for the MQTT server. If specified a valid certificate for the server must also be provided.'"#
+        )
+        .requires("server-certificate"))
+        .arg(Arg::from_usage(
+r#"--server-certificate=[FILE]   'Path to the TLS public certificate for the MQTT server. If specified a valid private key for the server must also be provided.'"#
+        )
+        .requires("server-private-key"))
+        .arg(Arg::from_usage(
+r#"--root-ca-certificate=[FILE]   'Path to the certificate of the certificate authority used to validate clients connecting to the MQTT server. If specified a valid private key and certificate for the server must also be provided.'"#
+        )
+        .requires_all(&["server-certificate", "server-private-key"]));
     let args = app.get_matches();
 
     // load config file at first
@@ -163,6 +175,9 @@ r#"-w, --generalise-pub=[String]...   'A list of key expression to use for gener
     insert_json5!(config, args, "plugins/mqtt/deny", if "deny", );
     insert_json5!(config, args, "plugins/mqtt/generalise_pubs", for "generalise-pub", .collect::<Vec<_>>());
     insert_json5!(config, args, "plugins/mqtt/generalise_subs", for "generalise-sub", .collect::<Vec<_>>());
+    insert_json5!(config, args, "plugins/mqtt/tls/server_private_key", if "server-private-key", );
+    insert_json5!(config, args, "plugins/mqtt/tls/server_certificate", if "server-certificate", );
+    insert_json5!(config, args, "plugins/mqtt/tls/root_ca_certificate", if "root-ca-certificate", );
     config
 }
 

zenoh-plugin-mqtt/Cargo.toml

@@ -35,6 +35,7 @@ stats = ["zenoh/stats"]
 async-channel = { workspace = true }
 async-std = { workspace = true, features = ["unstable", "attributes"] }
 async-trait = { workspace = true }
+base64 = { workspace = true }
 derivative = { workspace = true }
 env_logger = { workspace = true }
 flume = { workspace = true }
@@ -45,7 +46,11 @@ lazy_static = { workspace = true }
 log = { workspace = true }
 ntex = { workspace = true, features = ["async-std", "rustls"] }
 ntex-mqtt = { workspace = true }
+ntex-tls = { workspace = true }
 regex = { workspace = true }
+rustls = { workspace = true }
+rustls-pemfile = { workspace = true }
+secrecy = {workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 zenoh = { workspace = true }

zenoh-plugin-mqtt/src/config.rs

@@ -15,6 +15,7 @@ use regex::Regex;
 use serde::de::{Unexpected, Visitor};
 use serde::{de, Deserialize, Deserializer, Serialize, Serializer};
 use std::fmt;
+use zenoh::config::SecretValue;
 use zenoh::prelude::*;
 
 const DEFAULT_MQTT_INTERFACE: &str = "0.0.0.0";
@@ -46,11 +47,27 @@ pub struct Config {
     pub generalise_subs: Vec<OwnedKeyExpr>,
     #[serde(default)]
     pub generalise_pubs: Vec<OwnedKeyExpr>,
+    #[serde(default)]
+    pub tls: Option<TLSConfig>,
     __required__: Option<bool>,
     #[serde(default, deserialize_with = "deserialize_path")]
     __path__: Option<Vec<String>>,
 }
 
+#[derive(Deserialize, Serialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
+pub struct TLSConfig {
+    pub server_private_key: Option<String>,
+    #[serde(skip_serializing)]
+    pub server_private_key_base64: Option<SecretValue>,
+    pub server_certificate: Option<String>,
+    #[serde(skip_serializing)]
+    pub server_certificate_base64: Option<SecretValue>,
+    pub root_ca_certificate: Option<String>,
+    #[serde(skip_serializing)]
+    pub root_ca_certificate_base64: Option<SecretValue>,
+}
+
 fn default_mqtt_port() -> String {
     format!("{DEFAULT_MQTT_INTERFACE}:{DEFAULT_MQTT_PORT}")
 }