feat: add acl on topics

Author: SharksT

src/cli/run.rs

@@ -10,6 +10,7 @@ use tokio_rustls::rustls;
 
 use crate::cli::LogFormat;
 use crate::config;
+use crate::config::acl::AclConfig;
 use crate::config::addresses::Addresses;
 use crate::config::users::{AuthConfig, UsersConfig};
 use crate::mqtt::broker::{self, MqttBroker};
@@ -135,6 +136,7 @@ impl SecretKeyOpt {
 
 pub fn main(args: RunArgs) -> crate::Result<()> {
     let mut users = config::users::read(&args.config_dir.join("users.toml"))?;
+    let acl = config::acl::read(&args.config_dir.join("acl.toml"))?;
 
     // Merge any auth overrides from the command-line.
     users.auth.merge(&args.auth_config);
@@ -214,14 +216,15 @@ pub fn main(args: RunArgs) -> crate::Result<()> {
 
     let ws_config = args.ws_config.websockets.then(|| args.ws_config.clone());
 
-    main_async(args, users, tce_config, tls_config, ws_config)
+    main_async(args, users, acl, tce_config, tls_config, ws_config)
 }
 
 // `#[tokio::main]` doesn't have to be attached to the actual `main()`, and it can accept args
 #[tokio::main]
 async fn main_async(
     args: RunArgs,
     users: UsersConfig,
+    acl: AclConfig,
     tce_config: Option<tashi_consensus_engine::Config>,
     tls_config: Option<broker::TlsConfig>,
     ws_config: Option<WsConfig>,
@@ -244,6 +247,7 @@ async fn main_async(
         tls_config,
         ws_config,
         users,
+        acl,
         tce_platform.clone(),
         tce_message_stream,
         KeepAlive::from_seconds(args.max_keep_alive),

src/config/acl.rs

@@ -0,0 +1,62 @@
+use std::path::Path;
+
+use tashi_collections::HashMap;
+
+use crate::mqtt::trie::Filter;
+
+#[derive(serde::Deserialize, serde::Serialize, Default)]
+pub struct AclConfig {
+    pub permissions: HashMap<String, TopicsConfig>,
+}
+
+#[derive(serde::Deserialize, serde::Serialize)]
+pub struct TopicsConfig {
+    pub topic: Vec<TopicPermissions>,
+}
+
+#[derive(serde::Deserialize, serde::Serialize)]
+pub struct TopicPermissions {
+    pub filter: String,
+    pub allowed: Vec<TransactionType>,
+    pub denied: Vec<TransactionType>,
+}
+
+#[derive(serde::Deserialize, serde::Serialize, PartialEq, Eq)]
+pub enum TransactionType {
+    Subscribe,
+    Publish,
+}
+
+impl AclConfig {
+    pub fn get_topics_acl_config(&self, user: &str) -> Option<&TopicsConfig> {
+        match self.permissions.get(user) {
+            Some(permission) => Some(permission),
+            None => self.permissions.get("*"),
+        }
+    }
+
+    pub fn check_acl_config(
+        &self,
+        topics_config: Option<&TopicsConfig>,
+        filter: &Filter,
+        transaction_type: TransactionType,
+    ) -> bool {
+        // Allows everything if no topics config was found.
+        topics_config.map_or(true, |perms| {
+            !perms.topic.iter().any(|k| {
+                k.allowed.iter().any(|k| *k == transaction_type) && filter.matches_topic(&k.filter)
+            })
+        })
+    }
+}
+
+pub fn read(path: &Path) -> crate::Result<AclConfig> {
+    Ok(super::read_toml_optional("acl", path)?.unwrap_or_else(|| {
+        tracing::debug!(
+            "acl file not found at {}; any user can do anything with the topics.",
+            path.display()
+        );
+
+        AclConfig::default()
+    }))
+}

src/config/mod.rs

@@ -3,8 +3,8 @@ use serde::de::DeserializeOwned;
 use std::path::Path;
 use std::{fs, io};
 
+pub mod acl;
 pub mod addresses;
-
 pub mod users;
 
 fn read_toml<T: DeserializeOwned>(name: &str, path: &Path) -> crate::Result<T> {

src/mqtt/broker/connection.rs

@@ -461,6 +461,8 @@ impl<S: MqttSocket> Connection<S> {
 
         self.protocol = protocol;
 
+        let mut user = "".to_string();
+
         if connect_props.as_ref().is_some_and(|props| {
             props.authentication_method.is_some() || props.authentication_data.is_some()
         }) {
@@ -487,17 +489,16 @@ impl<S: MqttSocket> Connection<S> {
         };
 
         if let Some(login) = login {
-            let Some(user) = self.shared.users.by_username.get(&login.username) else {
+            let Some(logged_user) = self.shared.users.by_username.get(&login.username) else {
                 self.disconnect_on_connect_error(ConnectReturnCode::NotAuthorized, "unknown user")
                     .await?;
-
                 return Ok(None);
             };
 
             let verified = self
                 .shared
                 .password_hasher
-                .verify(login.password.as_bytes(), &user.password_hash)
+                .verify(login.password.as_bytes(), &logged_user.password_hash)
                 .await?;
 
             if !verified {
@@ -509,6 +510,8 @@ impl<S: MqttSocket> Connection<S> {
 
                 return Ok(None);
             }
+
+            user = login.username;
         } else if !self.shared.users.auth.allow_anonymous_login {
             self.disconnect_on_connect_error(
                 ConnectReturnCode::NotAuthorized,
@@ -592,6 +595,7 @@ impl<S: MqttSocket> Connection<S> {
                 self.id,
                 response.client_index,
                 client_id,
+                user,
                 store.mailbox.sender(),
                 clean_session,
             )

src/mqtt/broker/mod.rs

@@ -20,6 +20,7 @@ use connection::Connection;
 use rumqttd_protocol::QoS;
 
 use crate::cli::run::WsConfig;
+use crate::config::acl::AclConfig;
 use crate::config::users::UsersConfig;
 use crate::mqtt::broker::socket::{DirectSocket, MqttSocket};
 use crate::mqtt::broker::tls::TlsAcceptor;
@@ -175,6 +176,7 @@ impl MqttBroker {
         tls_config: Option<TlsConfig>,
         ws_config: Option<WsConfig>,
         users: UsersConfig,
+        acl: AclConfig,
         tce_platform: Option<Arc<Platform>>,
         tce_messages: Option<MessageStream>,
         max_keep_alive: KeepAlive,
@@ -205,7 +207,7 @@ impl MqttBroker {
 
         let (broker_tx, broker_rx) = mpsc::channel(100);
 
-        let router = MqttRouter::start(tce_platform.clone(), tce_messages, token.clone());
+        let router = MqttRouter::start(tce_platform.clone(), tce_messages, token.clone(), acl);
 
         Ok(MqttBroker {
             listen_addr,

src/mqtt/router.rs

@@ -2,6 +2,7 @@ use std::cmp;
 use std::collections::BTreeMap;
 use std::num::NonZeroU32;
 use std::ops::{Index, IndexMut};
+use std::str::FromStr;
 use std::sync::{Arc, OnceLock};
 use std::time::{Instant, SystemTime};
 
@@ -21,6 +22,7 @@ use tracing::Span;
 
 use rumqttd_protocol::{QoS, RetainForwardRule, SubscribeReasonCode, UnsubAckReason};
 
+use crate::config::acl::AclConfig;
 use crate::map_join_error;
 use crate::mqtt::mailbox::MailSender;
 use crate::mqtt::packets::PacketId;
@@ -72,6 +74,7 @@ impl MqttRouter {
         tce_platform: Option<Arc<Platform>>,
         tce_messages: Option<MessageStream>,
         token: CancellationToken,
+        acl: AclConfig,
     ) -> Self {
         let (command_tx, command_rx) = mpsc::channel(COMMAND_CAPACITY);
 
@@ -90,6 +93,7 @@ impl MqttRouter {
 
         let state = RouterState {
             token,
+            acl,
             clients: SecondaryMap::new(),
             dead_clients: HashSet::default(),
             subscriptions: Subscriptions::default(),
@@ -161,6 +165,7 @@ impl RouterHandle {
         connection_id: ConnectionId,
         client_index: ClientIndex,
         client_id: ClientId,
+        user: String,
         mail_tx: MailSender,
         clean_session: bool,
     ) -> crate::Result<RouterConnection> {
@@ -172,6 +177,7 @@ impl RouterHandle {
                 RouterCommand::NewConnection {
                     connection_id,
                     client_id,
+                    user,
                     message_tx,
                     mail_tx,
                     clean_session,
@@ -286,6 +292,7 @@ enum RouterCommand {
         connection_id: ConnectionId,
         client_id: ClientId,
         mail_tx: MailSender,
+        user: String,
         message_tx: mpsc::UnboundedSender<RouterMessage>,
         clean_session: bool,
     },
@@ -336,6 +343,8 @@ struct RouterState {
     clients: SecondaryMap<ClientIndex, ClientState>,
     dead_clients: HashSet<ClientIndex>,
 
+    acl: AclConfig,
+
     subscriptions: Subscriptions,
     command_rx: mpsc::Receiver<(ClientIndex, RouterCommand)>,
     system_rx: mpsc::UnboundedReceiver<SystemCommand>,
@@ -358,6 +367,7 @@ struct Tce {
 struct ClientState {
     client_id: ClientId,
     mail_tx: MailSender,
+    user: String,
     subscriptions: ClientSubscriptions,
     current_connection: Option<ConnectionState>,
     clean_session: bool,
@@ -421,6 +431,15 @@ enum PublishOrigin<'a> {
     Consensus(&'a CreatorId),
 }
 
+impl PublishOrigin<'_> {
+    pub fn get_client_index(&self) -> Option<ClientIndex> {
+        match self {
+            PublishOrigin::Local(client_index) => Some(*client_index),
+            _ => None,
+        }
+    }
+}
+
 impl Index<SubscriptionKind> for Subscriptions {
     type Output = SubscriptionMap;
 
@@ -565,6 +584,7 @@ fn handle_command(state: &mut RouterState, client_idx: ClientIndex, command: Rou
         RouterCommand::NewConnection {
             connection_id,
             client_id,
+            user,
             message_tx,
             mail_tx,
             clean_session,
@@ -591,6 +611,7 @@ fn handle_command(state: &mut RouterState, client_idx: ClientIndex, command: Rou
                 client_idx,
                 ClientState {
                     client_id,
+                    user,
                     mail_tx,
                     subscriptions: Default::default(),
                     current_connection: Some(ConnectionState {
@@ -657,9 +678,11 @@ fn handle_subscribe(state: &mut RouterState, client_idx: ClientIndex, request: S
         publish: Arc<PublishTrasaction>,
     }
 
-    if !state.clients.contains_key(client_idx) {
+    let Some(client) = state.clients.get(client_idx) else {
         return;
-    }
+    };
+
+    let permissions = state.acl.get_topics_acl_config(&client.user);
 
     // if state.connections[conn_id].message_tx.is_closed() {
     //     return;
@@ -677,6 +700,14 @@ fn handle_subscribe(state: &mut RouterState, client_idx: ClientIndex, request: S
                 // as they would have failed validation on the frontend.
                 .ok_or(SubscribeReasonCode::Unspecified)
                 .and_then(|(filter, props)| {
+                    if !state.acl.check_acl_config(
+                        permissions,
+                        &filter,
+                        crate::config::acl::TransactionType::Subscribe,
+                    ) {
+                        Err(SubscribeReasonCode::NotAuthorized)?
+                    }
+
                     let sub_kind = SubscriptionKind::from_filter(&filter)
                         .map_err(|_| SubscribeReasonCode::NotAuthorized)?;
 
@@ -904,6 +935,25 @@ fn dispatch(state: &mut RouterState, publish: Arc<PublishTrasaction>, origin: Pu
         },
     };
 
+    // Check if user has permission to publish
+    if let Some(client_index) = origin.get_client_index() {
+        let Some(client) = state.clients.get(client_index) else {
+            return;
+        };
+
+        let topic_filter = Filter::from(&topic);
+
+        let topics_config = state.acl.get_topics_acl_config(&client.user);
+
+        if !state.acl.check_acl_config(
+            topics_config,
+            &topic_filter,
+            crate::config::acl::TransactionType::Publish,
+        ) {
+            return;
+        }
+    }
+
     // Only run this if TCE is not available.
     if state.tce.is_none() && publish.meta.retain() {
         let time_now = state.startup_time + state.startup_instant.elapsed();