Add permissions (#55)

* feat: add acl on topics

* chore: serialize as filter directly from serde, fix check_acl_config

* chore: add test

* fix: tests

* chore: rename acl to permissions

Author: SharksT

.gitignore

@@ -6,3 +6,4 @@ foxmq.d/
 
 # These are test keys and must not be used anywhere else.
 !/tests/dmq
+.env

src/cli/run.rs

@@ -6,6 +6,7 @@ use tashi_collections::HashMap;
 use crate::cli::LogFormat;
 use crate::config;
 use crate::config::addresses::Addresses;
+use crate::config::permissions::PermissionsConfig;
 use crate::config::users::{AuthConfig, UsersConfig};
 use crate::mqtt::broker::{self, MqttBroker};
 use crate::mqtt::{KeepAlive, TceState};
@@ -181,6 +182,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::permissions::read(&args.config_dir.join("permissions.toml"))?;
 
     // Merge any auth overrides from the command-line.
     users.auth.merge(&args.auth_config);
@@ -261,14 +263,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,
+    permissions_config: PermissionsConfig,
     tce_config: Option<TceConfig>,
     tls_config: Option<broker::TlsConfig>,
     ws_config: Option<WsConfig>,
@@ -298,6 +301,7 @@ async fn main_async(
         tls_config,
         ws_config,
         users,
+        permissions_config,
         tce,
         KeepAlive::from_seconds(args.max_keep_alive),
     )

src/config/mod.rs

@@ -4,7 +4,7 @@ use std::path::Path;
 use std::{fs, io};
 
 pub mod addresses;
-
+pub mod permissions;
 pub mod users;
 
 fn read_toml<T: DeserializeOwned>(name: &str, path: &Path) -> crate::Result<T> {

src/config/permissions.rs

@@ -0,0 +1,82 @@
+use std::{path::Path, str::FromStr};
+
+use tashi_collections::HashMap;
+
+use crate::mqtt::trie::Filter;
+
+#[derive(serde::Deserialize, Default)]
+pub struct PermissionsConfig {
+    #[serde(default)]
+    pub permissions: HashMap<String, TopicsConfig>,
+}
+
+#[derive(serde::Deserialize, Debug)]
+pub struct TopicsConfig {
+    pub topic: Vec<TopicPermissions>,
+}
+
+#[derive(serde::Deserialize, Debug)]
+pub struct TopicPermissions {
+    #[serde(deserialize_with = "from_str")]
+    pub filter: Filter,
+    pub allowed: Vec<TransactionType>,
+    #[serde(default)]
+    pub denied: Vec<TransactionType>,
+}
+
+fn from_str<'de, D>(deserializer: D) -> Result<Filter, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let s: String = serde::Deserialize::deserialize(deserializer)?;
+
+    Filter::from_str(&s).map_err(serde::de::Error::custom)
+}
+
+#[derive(serde::Deserialize, PartialEq, Eq, Debug)]
+#[serde(rename_all = "lowercase")]
+pub enum TransactionType {
+    Subscribe,
+    Publish,
+}
+
+impl PermissionsConfig {
+    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>,
+        topic_name: &str,
+        transaction_type: TransactionType,
+    ) -> bool {
+        // Allows everything if no topics config was found.
+        topics_config.map_or(true, |perms| {
+            perms
+                .topic
+                .iter()
+                .find(|k| k.filter.matches_topic(topic_name))
+                .map_or(true, |k| {
+                    k.allowed.iter().any(|k| *k == transaction_type)
+                        || !k.denied.iter().all(|k| *k == transaction_type)
+                })
+        })
+    }
+}
+
+pub fn read(path: &Path) -> crate::Result<PermissionsConfig> {
+    Ok(
+        super::read_toml_optional("permissions", path)?.unwrap_or_else(|| {
+            tracing::debug!(
+                "permissions file not found at {}; any user can do anything with the topics.",
+                path.display()
+            );
+
+            PermissionsConfig::default()
+        }),
+    )
+}

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

@@ -21,6 +21,7 @@ use connection::Connection;
 use rumqttd_protocol::QoS;
 
 use crate::cli::run::WsConfig;
+use crate::config::permissions::PermissionsConfig;
 use crate::config::users::UsersConfig;
 use crate::mqtt::broker::socket::{DirectSocket, MqttSocket};
 use crate::mqtt::broker::tls::TlsAcceptor;
@@ -171,11 +172,13 @@ pub struct TlsConfig {
 }
 
 impl MqttBroker {
+    #[allow(clippy::too_many_arguments)]
     pub async fn bind(
         listen_addr: SocketAddr,
         tls_config: Option<TlsConfig>,
         ws_config: Option<WsConfig>,
         users: UsersConfig,
+        permissions_config: PermissionsConfig,
         tce: Option<TceState>,
         max_keep_alive: KeepAlive,
     ) -> crate::Result<Self> {
@@ -207,7 +210,7 @@ impl MqttBroker {
 
         let tce_platform = tce.as_ref().map(|tce| tce.platform.clone());
 
-        let router = MqttRouter::start(tce, token.clone());
+        let router = MqttRouter::start(tce, token.clone(), permissions_config);
 
         Ok(MqttBroker {
             listen_addr,

src/mqtt/router.rs

@@ -23,6 +23,7 @@ use tracing::{Instrument, Span};
 
 use rumqttd_protocol::{QoS, RetainForwardRule, SubscribeReasonCode, UnsubAckReason};
 
+use crate::config::permissions::PermissionsConfig;
 use crate::map_join_error;
 use crate::mqtt::mailbox::MailSender;
 use crate::mqtt::packets::PacketId;
@@ -77,7 +78,11 @@ pub struct TceState {
 }
 
 impl MqttRouter {
-    pub fn start(tce: Option<TceState>, token: CancellationToken) -> Self {
+    pub fn start(
+        tce: Option<TceState>,
+        token: CancellationToken,
+        permissions: PermissionsConfig,
+    ) -> Self {
         let (command_tx, command_rx) = mpsc::channel(COMMAND_CAPACITY);
 
         let (system_tx, system_rx) = mpsc::unbounded_channel();
@@ -88,6 +93,7 @@ impl MqttRouter {
 
         let state = RouterState {
             token,
+            permissions,
             clients: SecondaryMap::new(),
             dead_clients: HashSet::default(),
             subscriptions: Subscriptions::default(),
@@ -159,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> {
@@ -170,6 +177,7 @@ impl RouterHandle {
                 RouterCommand::NewConnection {
                     connection_id,
                     client_id,
+                    user,
                     message_tx,
                     mail_tx,
                     clean_session,
@@ -284,6 +292,7 @@ enum RouterCommand {
         connection_id: ConnectionId,
         client_id: ClientId,
         mail_tx: MailSender,
+        user: String,
         message_tx: mpsc::UnboundedSender<RouterMessage>,
         clean_session: bool,
     },
@@ -334,6 +343,8 @@ struct RouterState {
     clients: SecondaryMap<ClientIndex, ClientState>,
     dead_clients: HashSet<ClientIndex>,
 
+    permissions: PermissionsConfig,
+
     subscriptions: Subscriptions,
     command_rx: mpsc::Receiver<(ClientIndex, RouterCommand)>,
     system_rx: mpsc::UnboundedReceiver<SystemCommand>,
@@ -351,6 +362,7 @@ struct RouterState {
 struct ClientState {
     client_id: ClientId,
     mail_tx: MailSender,
+    user: String,
     subscriptions: ClientSubscriptions,
     current_connection: Option<ConnectionState>,
     clean_session: bool,
@@ -414,6 +426,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;
 
@@ -567,6 +588,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,
@@ -593,6 +615,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 {
@@ -663,9 +686,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.permissions.get_topics_acl_config(&client.user);
 
     // if state.connections[conn_id].message_tx.is_closed() {
     //     return;
@@ -683,6 +708,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.permissions.check_acl_config(
+                        permissions,
+                        filter.as_str(),
+                        crate::config::permissions::TransactionType::Subscribe,
+                    ) {
+                        Err(SubscribeReasonCode::NotAuthorized)?
+                    }
+
                     let sub_kind = SubscriptionKind::from_filter(&filter)
                         .map_err(|_| SubscribeReasonCode::NotAuthorized)?;
 
@@ -974,6 +1007,23 @@ 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 topics_config = state.permissions.get_topics_acl_config(&client.user);
+
+        if !state.permissions.check_acl_config(
+            topics_config,
+            &publish.topic,
+            crate::config::permissions::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();

tests/foxmq.d/permissions.toml

@@ -0,0 +1,6 @@
+[permissions.test_user1]
+topic = [{ filter = "test_topic", allowed = ["publish"] }]
+
+
+[permissions.test_user2]
+topic = [{ filter = "test_topic", allowed = ["subscribe"] }]

tests/foxmq.d/users.toml

@@ -1,2 +1,8 @@
 [auth]
 allow-anonymous-login = true
+
+[users.test_user1]
+password-hash = "$argon2id$v=19$m=19456,t=2,p=1$TwFLGuZeIBQGjn02H9NpJQ$8s2JrmbxhLfHphYEfmZ4KjJmxS7tTCss06E27kZC6M0"
+
+[users.test_user2]
+password-hash = "$argon2id$v=19$m=19456,t=2,p=1$TwFLGuZeIBQGjn02H9NpJQ$8s2JrmbxhLfHphYEfmZ4KjJmxS7tTCss06E27kZC6M0"