feat: Support for multi tenancy prefix checks (bytebeamio#505)

* Enable multitenancy again

* Update cargo lock

* Add feature flag for tenant prefix validation

* Fix shadow compilation error

* Cargo format

* Don't error on empty organisation in TLS cert

* Update changelog

* Update stale parts of main branch

* Remove meters link which is included by mistake (not included in module tree anyway)

* Default to not having multitenancy prefix checks

Co-authored-by: Mrinal Paliwal <mrinal16164@iiitd.ac.in>

CHANGELOG.md

@@ -8,6 +8,8 @@ rumqttc
 
 rumqttd
 -------
+- Add meters related to router, subscriptions, and connections (#505)
+- Allow multi-tenancy validation for mtls clients with `Org` set in certificates
 - Add `tracing` for structured, context-aware logging (#499, #503)
 - Added properties field to `Unsubscribe`, `UnsubAck`, and `Disconnect` packets so its consistent with other packets. (#480)
 - Changed default segment size in demo config to 100MB (#484)

rumqttd/Cargo.toml

@@ -24,7 +24,7 @@ rustls-pemfile = { version = "0.3.0", optional = true }
 tokio-tungstenite = { version = "0.15.0", optional = true }
 websocket-codec = { version = "0.5.1", optional = true }
 rouille = "3.1.1"
-# x509-parser = {version= "0.9.2", optional = true}
+x509-parser = {version= "0.9.2", optional = true}
 futures-util = { version = "0.3.16", optional = true}
 parking_lot = "0.11.2"
 config = "0.13"
@@ -34,9 +34,10 @@ tracing-subscriber = { version="0.3.16", features=["env-filter"] }
 
 [features]
 default = ["use-rustls"]
-use-rustls = ["tokio-rustls", "rustls-pemfile"] #, "x509-parser"]
-use-native-tls = ["tokio-native-tls"] #, "x509-parser"]
+use-rustls = ["tokio-rustls", "rustls-pemfile", "x509-parser"]
+use-native-tls = ["tokio-native-tls" , "x509-parser"]
 websockets = ["tokio-tungstenite", "websocket-codec", "tokio-util", "futures-util"]
+validate-tenant-prefix = []
 
 [dev-dependencies]
 pretty_env_logger = "0.4.0"

rumqttd/src/link/console.rs

@@ -18,8 +18,7 @@ impl ConsoleLink {
     /// Requires the corresponding Router to be running to complete
     pub fn new(config: ConsoleSettings, router_tx: Sender<(ConnectionId, Event)>) -> ConsoleLink {
         let tx = router_tx.clone();
-        let (link_tx, link_rx, _ack) =
-            Link::new(/*None,*/ "console", tx, true, None, true).unwrap();
+        let (link_tx, link_rx, _ack) = Link::new(None, "console", tx, true, None, true).unwrap();
         let connection_id = link_tx.connection_id;
         ConsoleLink {
             config,

rumqttd/src/link/local.rs

@@ -40,7 +40,7 @@ pub struct Link;
 impl Link {
     #[allow(clippy::type_complexity)]
     fn prepare(
-        // tenant_id: Option<String>,
+        tenant_id: Option<String>,
         client_id: &str,
         clean: bool,
         last_will: Option<LastWill>,
@@ -53,7 +53,7 @@ impl Link {
         Receiver<MetricsReply>,
     ) {
         let (connection, metrics_rx) = Connection::new(
-            // tenant_id,
+            tenant_id,
             client_id.to_owned(),
             clean,
             last_will,
@@ -81,7 +81,7 @@ impl Link {
 
     #[allow(clippy::new_ret_no_self)]
     pub fn new(
-        // tenant_id: Option<String>,
+        tenant_id: Option<String>,
         client_id: &str,
         router_tx: Sender<(ConnectionId, Event)>,
         clean: bool,
@@ -91,12 +91,8 @@ impl Link {
         // Connect to router
         // Local connections to the router shall have access to all subscriptions
 
-        let (message, i, o, link_rx, metrics_rx) = Link::prepare(
-            /*tenant_id,*/ client_id,
-            clean,
-            last_will,
-            dynamic_filters,
-        );
+        let (message, i, o, link_rx, metrics_rx) =
+            Link::prepare(tenant_id, client_id, clean, last_will, dynamic_filters);
         router_tx.send((0, message))?;
 
         link_rx.recv()?;
@@ -115,7 +111,7 @@ impl Link {
     }
 
     pub async fn init(
-        // tenant_id: Option<String>,
+        tenant_id: Option<String>,
         client_id: &str,
         router_tx: Sender<(ConnectionId, Event)>,
         clean: bool,
@@ -125,12 +121,8 @@ impl Link {
         // Connect to router
         // Local connections to the router shall have access to all subscriptions
 
-        let (message, i, o, link_rx, metrics_rx) = Link::prepare(
-            /*tenant_id,*/ client_id,
-            clean,
-            last_will,
-            dynamic_filters,
-        );
+        let (message, i, o, link_rx, metrics_rx) =
+            Link::prepare(tenant_id, client_id, clean, last_will, dynamic_filters);
         router_tx.send_async((0, message)).await?;
 
         link_rx.recv_async().await?;

rumqttd/src/link/remote.rs

@@ -57,7 +57,7 @@ impl<P: Protocol> RemoteLink<P> {
     pub async fn new(
         config: Arc<ConnectionSettings>,
         router_tx: Sender<(ConnectionId, Event)>,
-        // tenant_id: Option<String>,
+        tenant_id: Option<String>,
         mut network: Network<P>,
     ) -> Result<RemoteLink<P>, Error> {
         // Wait for MQTT connect packet and error out if it's not received in time to prevent
@@ -91,7 +91,7 @@ impl<P: Protocol> RemoteLink<P> {
         }
 
         let (link_tx, link_rx, notification) = Link::new(
-            // tenant_id,
+            tenant_id,
             &client_id,
             router_tx,
             clean_session,