Updates: coderabbit suggestions

parmesant · parmesant · commit 8d6c65d87b05 · 2025-07-07T21:47:02.000+05:30
diff --git a/src/handlers/http/modal/ingest/ingestor_rbac.rs b/src/handlers/http/modal/ingest/ingestor_rbac.rs
@@ -24,6 +24,7 @@ use tokio::sync::Mutex;
 use crate::{
     handlers::http::{modal::utils::rbac_utils::get_metadata, rbac::RBACError},
     rbac::{
+        map::roles,
         user::{self, User as ParseableUser},
         Users,
     },
@@ -73,38 +74,6 @@ pub async fn delete_user(username: web::Path<String>) -> Result<impl Responder,
     Ok(format!("deleted user: {username}"))
 }
 
-// // Handler PUT /user/{username}/roles => Put roles for user
-// // Put roles for given user
-// pub async fn put_role(
-//     username: web::Path<String>,
-//     role: web::Json<HashSet<String>>,
-// ) -> Result<String, RBACError> {
-//     let username = username.into_inner();
-//     let role = role.into_inner();
-
-//     if !Users.contains(&username) {
-//         return Err(RBACError::UserDoesNotExist);
-//     };
-//     // update parseable.json first
-//     let mut metadata = get_metadata().await?;
-//     if let Some(user) = metadata
-//         .users
-//         .iter_mut()
-//         .find(|user| user.username() == username)
-//     {
-//         user.roles.clone_from(&role);
-//     } else {
-//         // should be unreachable given state is always consistent
-//         return Err(RBACError::UserDoesNotExist);
-//     }
-
-//     let _ = storage::put_staging_metadata(&metadata);
-//     // update in mem table
-//     Users.add_roles(&username.clone(), role.clone());
-
-//     Ok(format!("Roles updated successfully for {username}"))
-// }
-
 // Handler PATCH /user/{username}/role/sync/add => Add roles to a user
 pub async fn add_roles_to_user(
     username: web::Path<String>,
@@ -147,6 +116,30 @@ pub async fn remove_roles_from_user(
     if !Users.contains(&username) {
         return Err(RBACError::UserDoesNotExist);
     };
+
+    // check if all roles exist
+    let mut non_existent_roles = Vec::new();
+    roles_to_remove.iter().for_each(|r| {
+        if roles().get(r).is_none() {
+            non_existent_roles.push(r.clone());
+        }
+    });
+
+    if !non_existent_roles.is_empty() {
+        return Err(RBACError::RolesDoNotExist(non_existent_roles));
+    }
+
+    // check that user actually has these roles
+    let user_roles: HashSet<String> = HashSet::from_iter(Users.get_role(&username));
+    let roles_not_with_user: HashSet<String> =
+        HashSet::from_iter(roles_to_remove.difference(&user_roles).cloned());
+
+    if !roles_not_with_user.is_empty() {
+        return Err(RBACError::RolesNotAssigned(Vec::from_iter(
+            roles_not_with_user,
+        )));
+    }
+
     // update parseable.json first
     let mut metadata = get_metadata().await?;
     if let Some(user) = metadata
diff --git a/src/handlers/http/role.rs b/src/handlers/http/role.rs
@@ -16,8 +16,6 @@
  *
  */
 
-use std::collections::HashSet;
-
 use actix_web::{
     http::header::ContentType,
     web::{self, Json},
@@ -103,23 +101,21 @@ pub async fn delete(name: web::Path<String>) -> Result<impl Responder, RoleError
         };
     }
 
+    // remove role from all user groups that have it
     let mut groups_to_update = Vec::new();
-    for user_group in group_names {
-        if let Some(ug) = write_user_groups().get_mut(&user_group) {
-            ug.remove_roles(HashSet::from_iter([name.clone()]))
-                .map_err(|e| RoleError::Anyhow(anyhow::Error::msg(e.to_string())))?;
-            groups_to_update.push(ug.clone());
-            // ug.update_in_metadata().await?;
-        } else {
-            continue;
-        };
+    for group in write_user_groups().values_mut() {
+        if group.roles.remove(&name) {
+            groups_to_update.push(group.clone());
+        }
     }
 
-    // update in metadata
-    metadata
-        .user_groups
-        .retain(|x| !groups_to_update.contains(x));
-    metadata.user_groups.extend(groups_to_update);
+    // update metadata only if there are changes
+    if !groups_to_update.is_empty() {
+        metadata
+            .user_groups
+            .retain(|x| !groups_to_update.contains(x));
+        metadata.user_groups.extend(groups_to_update);
+    }
     put_metadata(&metadata).await?;
 
     Ok(HttpResponse::Ok().finish())
diff --git a/src/migration/mod.rs b/src/migration/mod.rs
@@ -41,6 +41,13 @@ use crate::{
     },
 };
 
+fn get_version(metadata: &serde_json::Value) -> Option<&str> {
+    metadata
+        .as_object()
+        .and_then(|meta| meta.get("version"))
+        .and_then(|version| version.as_str())
+}
+
 /// Migrate the metdata from v1 or v2 to v3
 /// This is a one time migration
 pub async fn run_metadata_migration(
@@ -55,13 +62,6 @@ pub async fn run_metadata_migration(
     }
     let staging_metadata = get_staging_metadata(config)?;
 
-    fn get_version(metadata: &serde_json::Value) -> Option<&str> {
-        metadata
-            .as_object()
-            .and_then(|meta| meta.get("version"))
-            .and_then(|version| version.as_str())
-    }
-
     // if storage metadata is none do nothing
     if let Some(storage_metadata) = storage_metadata {
         match get_version(&storage_metadata) {
@@ -117,37 +117,42 @@ pub async fn run_metadata_migration(
 
     // if staging metadata is none do nothing
     if let Some(staging_metadata) = staging_metadata {
-        match get_version(&staging_metadata) {
-            Some("v1") => {
-                let mut metadata = metadata_migration::v1_v3(staging_metadata);
-                metadata = metadata_migration::v3_v4(metadata);
-                put_staging_metadata(config, &metadata)?;
-            }
-            Some("v2") => {
-                let mut metadata = metadata_migration::v2_v3(staging_metadata);
-                metadata = metadata_migration::v3_v4(metadata);
-                put_staging_metadata(config, &metadata)?;
-            }
-            Some("v3") => {
-                let metadata = metadata_migration::v3_v4(staging_metadata);
-                put_staging_metadata(config, &metadata)?;
-            }
-            Some("v4") => {
-                let metadata = metadata_migration::v4_v5(staging_metadata);
-                let metadata = metadata_migration::v5_v6(metadata);
-                put_staging_metadata(config, &metadata)?;
-            }
-            Some("v5") => {
-                let metadata = metadata_migration::v5_v6(staging_metadata);
-                put_staging_metadata(config, &metadata)?;
-            }
-            _ => (),
-        }
+        migrate_staging(config, staging_metadata)?;
     }
 
     Ok(())
 }
 
+fn migrate_staging(config: &Parseable, staging_metadata: Value) -> anyhow::Result<()> {
+    match get_version(&staging_metadata) {
+        Some("v1") => {
+            let mut metadata = metadata_migration::v1_v3(staging_metadata);
+            metadata = metadata_migration::v3_v4(metadata);
+            put_staging_metadata(config, &metadata)?;
+        }
+        Some("v2") => {
+            let mut metadata = metadata_migration::v2_v3(staging_metadata);
+            metadata = metadata_migration::v3_v4(metadata);
+            put_staging_metadata(config, &metadata)?;
+        }
+        Some("v3") => {
+            let metadata = metadata_migration::v3_v4(staging_metadata);
+            put_staging_metadata(config, &metadata)?;
+        }
+        Some("v4") => {
+            let metadata = metadata_migration::v4_v5(staging_metadata);
+            let metadata = metadata_migration::v5_v6(metadata);
+            put_staging_metadata(config, &metadata)?;
+        }
+        Some("v5") => {
+            let metadata = metadata_migration::v5_v6(staging_metadata);
+            put_staging_metadata(config, &metadata)?;
+        }
+        _ => (),
+    }
+    Ok(())
+}
+
 /// run the migration for all streams concurrently
 pub async fn run_migration(config: &Parseable) -> anyhow::Result<()> {
     let storage = config.storage.get_object_store();
diff --git a/src/rbac/map.rs b/src/rbac/map.rs
@@ -226,33 +226,9 @@ impl Sessions {
         context_user: Option<&str>,
     ) -> Option<Response> {
         self.active_sessions.get(key).map(|(username, perms)| {
-            // if user is a part of any user groups, then add permissions
-            let perms: HashSet<Permission> = if let Some(user) = users().0.get(username) {
-                let all_groups_roles = user
-                    .user_groups
-                    .iter()
-                    .filter(|id| (read_user_groups().0.contains_key(*id)))
-                    .map(|id| read_user_groups().0.get(id).unwrap().roles.clone())
-                    .reduce(|mut acc, e| {
-                        acc.extend(e);
-                        acc
-                    })
-                    .unwrap_or_default();
-
-                let mut privilege_list = Vec::new();
-                all_groups_roles
-                    .iter()
-                    .filter_map(|role| roles().get(role).cloned())
-                    .for_each(|privileges| privilege_list.extend(privileges));
-
-                let mut perms = HashSet::from_iter(perms.clone());
-                for privs in privilege_list {
-                    perms.extend(RoleBuilder::from(&privs).build())
-                }
-                perms
-            } else {
-                HashSet::from_iter(perms.clone())
-            };
+            let mut perms: HashSet<Permission> = HashSet::from_iter(perms.clone());
+            perms.extend(aggregate_group_permissions(username));
+
             if perms.iter().any(|user_perm| {
                 match *user_perm {
                     // if any action is ALL then we we authorize
@@ -317,6 +293,35 @@ impl From<Vec<User>> for Users {
     }
 }
 
+fn aggregate_group_permissions(username: &str) -> HashSet<Permission> {
+    let mut group_perms = HashSet::new();
+
+    let Some(user) = users().get(username).cloned() else {
+        return group_perms;
+    };
+
+    if user.user_groups.is_empty() {
+        return group_perms;
+    }
+
+    for group_name in &user.user_groups {
+        let Some(group) = read_user_groups().get(group_name).cloned() else {
+            continue;
+        };
+
+        for role_name in &group.roles {
+            let Some(privileges) = roles().get(role_name).cloned() else {
+                continue;
+            };
+
+            for privilege in privileges {
+                group_perms.extend(RoleBuilder::from(&privilege).build());
+            }
+        }
+    }
+
+    group_perms
+}
 // Map of [user group ID --> UserGroup]
 // This map is populated at startup with the list of user groups from parseable.json file
 #[derive(Debug, Default, Clone, derive_more::Deref, derive_more::DerefMut)]
diff --git a/src/rbac/role.rs b/src/rbac/role.rs
