feat(mcp): add resource group scopes with permission inheritance

crates/dbx-core/src/mcp_policy.rs

@@ -1,10 +1,107 @@
+use std::collections::HashMap;
+
 use crate::models::connection::ConnectionConfig;
 use crate::production_safety::sql_references_disallowed_database;
 use crate::storage::McpGlobalPolicy;
+use serde::Deserialize;
+use serde_json::Value;
 
 /// Version used by rules that opt into scoped override semantics.
 pub const MCP_EXECUTION_POLICY_VERSION: u8 = 1;
 
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct McpConnectionGroupPath {
+    pub ids: Vec<String>,
+    pub names: Vec<String>,
+}
+
+#[derive(Deserialize)]
+struct SidebarLayout {
+    #[serde(default)]
+    groups: Vec<SidebarGroup>,
+    #[serde(default)]
+    order: Vec<SidebarOrderEntry>,
+}
+
+#[derive(Deserialize)]
+struct SidebarGroup {
+    id: String,
+    name: String,
+}
+
+#[derive(Deserialize)]
+#[serde(tag = "type")]
+enum SidebarOrderEntry {
+    #[serde(rename = "group")]
+    Group {
+        id: String,
+        children: Option<Vec<SidebarOrderEntry>>,
+        #[serde(rename = "connectionIds")]
+        connection_ids: Option<Vec<String>>,
+    },
+    #[serde(rename = "connection")]
+    Connection { id: String },
+}
+
+/// Resolve each connection's stable group ids and display names from the
+/// persisted sidebar tree. Legacy `connectionIds` group entries remain valid.
+pub fn connection_group_paths(layout: &Value) -> Result<HashMap<String, McpConnectionGroupPath>, String> {
+    let layout = serde_json::from_value::<SidebarLayout>(layout.clone())
+        .map_err(|error| format!("INVALID_SIDEBAR_LAYOUT: {error}"))?;
+    let groups = layout.groups.into_iter().map(|group| (group.id, group.name)).collect::<HashMap<_, _>>();
+    let mut paths = HashMap::new();
+    collect_connection_group_paths(&layout.order, &groups, &mut McpConnectionGroupPath::default(), &mut paths)?;
+    Ok(paths)
+}
+
+fn collect_connection_group_paths(
+    entries: &[SidebarOrderEntry],
+    groups: &HashMap<String, String>,
+    path: &mut McpConnectionGroupPath,
+    paths: &mut HashMap<String, McpConnectionGroupPath>,
+) -> Result<(), String> {
+    for entry in entries {
+        match entry {
+            SidebarOrderEntry::Connection { id } => {
+                paths.insert(id.clone(), path.clone());
+            }
+            SidebarOrderEntry::Group { id, children, connection_ids } => {
+                let name =
+                    groups.get(id).ok_or_else(|| format!("INVALID_SIDEBAR_LAYOUT: group '{id}' has no metadata"))?;
+                path.ids.push(id.clone());
+                path.names.push(name.clone());
+                if let Some(children) = children {
+                    collect_connection_group_paths(children, groups, path, paths)?;
+                } else if let Some(connection_ids) = connection_ids {
+                    for connection_id in connection_ids {
+                        paths.insert(connection_id.clone(), path.clone());
+                    }
+                }
+                path.ids.pop();
+                path.names.pop();
+            }
+        }
+    }
+    Ok(())
+}
+
+pub fn policy_uses_connection_groups(policy: &McpGlobalPolicy) -> bool {
+    !policy.allowed_group_ids.is_empty() || !policy.group_policies.is_empty()
+}
+
+pub fn policy_allows_connection(
+    policy: &McpGlobalPolicy,
+    group_path: Option<&McpConnectionGroupPath>,
+    connection_id: &str,
+) -> bool {
+    policy.allowed_connection_ids.as_ref().is_none_or(|allowed| {
+        allowed.iter().any(|id| id == connection_id)
+            || group_path.is_some_and(|path| {
+                path.ids.iter().any(|id| policy.allowed_group_ids.iter().any(|allowed| allowed == id))
+            })
+    })
+}
+
 /// Resolve an omitted or blank request database to the connection default.
 pub fn resolve_database(requested: &str, configured: Option<&str>) -> String {
     let requested = requested.trim();
@@ -22,8 +119,22 @@ pub fn effective_database_execution_policy(
     policy: &McpGlobalPolicy,
     connection_id: &str,
     database: &str,
+) -> (bool, bool) {
+    effective_database_execution_policy_with_groups(policy, &[], connection_id, database)
+}
+
+pub fn effective_database_execution_policy_with_groups(
+    policy: &McpGlobalPolicy,
+    group_ids: &[String],
+    connection_id: &str,
+    database: &str,
 ) -> (bool, bool) {
     let mut effective = (policy.read_only, policy.allow_dangerous_sql);
+    for group_id in group_ids {
+        if let Some(rule) = policy.group_policies.iter().find(|rule| rule.group_id == *group_id) {
+            effective = (rule.read_only, !rule.read_only && rule.allow_dangerous_sql);
+        }
+    }
     let Some(rule) = policy.connection_policies.iter().find(|rule| rule.connection_id == connection_id) else {
         return effective;
     };

crates/dbx-core/src/query.rs

@@ -6849,7 +6849,7 @@ for line in sys.stdin:
         let error = execute_sql_statement(&state, "conn-1", "", "SELECT 1", None, None).await.unwrap_err();
 
         assert!(error.contains("injected Agent failure"));
-        assert!(!state.pool_handle("conn-1").await.is_some());
+        assert!(state.pool_handle("conn-1").await.is_none());
         assert!(runtime.is_failed());
 
         runtime.kill();
@@ -6975,7 +6975,7 @@ for line in sys.stdin:
         .unwrap_err();
 
         assert!(error.contains("injected Agent failure"));
-        assert!(!state.pool_handle("conn-1").await.is_some());
+        assert!(state.pool_handle("conn-1").await.is_none());
         assert!(runtime.is_failed());
 
         runtime.kill();
@@ -6992,7 +6992,7 @@ for line in sys.stdin:
                 .unwrap_err();
 
         assert!(error.contains("injected Agent failure"));
-        assert!(!state.pool_handle("conn-1").await.is_some());
+        assert!(state.pool_handle("conn-1").await.is_none());
         assert!(runtime.is_failed());
 
         runtime.kill();
@@ -7006,7 +7006,7 @@ for line in sys.stdin:
         let error = execute_sql_statement(&state, "conn-1", "", "SELECT 1", None, None).await.unwrap_err();
 
         assert!(error.contains("injected Agent failure"));
-        assert!(!state.pool_handle("conn-1").await.is_some());
+        assert!(state.pool_handle("conn-1").await.is_none());
         assert!(!runtime.is_failed());
 
         runtime.kill();
@@ -8550,7 +8550,7 @@ for line in sys.stdin:
             execute_in_manual_transaction(&state, "txn-test", "SELECT 42", "dbx_test", None, Some(10)).await.unwrap();
         assert_eq!(results[0].rows, vec![vec![serde_json::json!(42)]]);
         commit_manual_transaction(&state, "txn-test").await.unwrap();
-        assert!(!state.pool_handle(pool_key).await.is_some());
+        assert!(state.pool_handle(pool_key).await.is_none());
         assert_eq!(
             std::fs::read_to_string(&calls).unwrap(),
             "beginManualTransaction\nexecuteInManualTransaction\ncommitManualTransaction\n"

crates/dbx-core/src/schema.rs

@@ -4097,7 +4097,7 @@ mod tests {
 
         replace_metadata_runtime(&state, "conn", Some("analytics"), None).await;
 
-        assert!(!state.pool_handle("conn:analytics:role:metadata").await.is_some());
+        assert!(state.pool_handle("conn:analytics:role:metadata").await.is_none());
         let _ = std::fs::remove_dir_all(dir);
     }
 
@@ -4126,7 +4126,7 @@ mod tests {
 
         assert_eq!(result.unwrap_err(), "Agent RPC call timed out (30s)");
         assert_eq!(attempts, 1);
-        assert!(!state.pool_handle("conn:role:metadata").await.is_some());
+        assert!(state.pool_handle("conn:role:metadata").await.is_none());
         let _ = std::fs::remove_dir_all(dir);
     }
 
@@ -4159,7 +4159,7 @@ mod tests {
 
         assert_eq!(result.unwrap_err(), quarantine);
         assert_eq!(attempts, 2);
-        assert!(!state.pool_handle("conn").await.is_some());
+        assert!(state.pool_handle("conn").await.is_none());
         let _ = std::fs::remove_dir_all(dir);
     }
 
@@ -4235,7 +4235,7 @@ for line in sys.stdin:
             Some(crate::db::agent_driver::AgentErrorCategory::Timeout)
         );
         assert_eq!(std::fs::read_to_string(call_count_path).unwrap(), "1");
-        assert!(!state.pool_handle(pool_key).await.is_some());
+        assert!(state.pool_handle(pool_key).await.is_none());
         runtime.kill();
         let _ = std::fs::remove_dir_all(dir);
     }

crates/dbx-core/src/storage.rs

@@ -234,6 +234,10 @@ pub struct McpGlobalPolicy {
     #[serde(default)]
     pub allow_dangerous_sql: bool,
     pub allowed_connection_ids: Option<Vec<String>>,
+    /// Stable sidebar group ids whose current descendant connections are
+    /// exposed when an explicit connection scope is configured.
+    #[serde(default)]
+    pub allowed_group_ids: Vec<String>,
     /// `None` exposes every built-in MCP tool. A list is an explicit
     /// allowlist and is enforced independently of connection permissions.
     #[serde(default)]
@@ -242,10 +246,25 @@ pub struct McpGlobalPolicy {
     /// the current execution policy version remain legacy ceilings.
     #[serde(default)]
     pub connection_policies: Vec<McpConnectionPolicy>,
+    /// Execution defaults inherited by every connection currently contained
+    /// in the referenced sidebar group. Nested groups are resolved from root
+    /// to leaf, so the closest configured group wins.
+    #[serde(default)]
+    pub group_policies: Vec<McpGroupPolicy>,
     #[serde(default)]
     pub query_timeout_secs: Option<u64>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct McpGroupPolicy {
+    pub group_id: String,
+    #[serde(default)]
+    pub read_only: bool,
+    #[serde(default)]
+    pub allow_dangerous_sql: bool,
+}
+
 fn default_mcp_connection_execution_mode_configured() -> bool {
     true
 }
@@ -368,10 +387,14 @@ pub struct McpGlobalPolicyState {
     pub allow_dangerous_sql: bool,
     pub allowed_connection_ids: Option<Vec<String>>,
     #[serde(default)]
+    pub allowed_group_ids: Vec<String>,
+    #[serde(default)]
     pub allowed_tool_names: Option<Vec<String>>,
     #[serde(default)]
     pub connection_policies: Vec<McpConnectionPolicy>,
     #[serde(default)]
+    pub group_policies: Vec<McpGroupPolicy>,
+    #[serde(default)]
     pub query_timeout_secs: Option<u64>,
 }
 
@@ -381,8 +404,10 @@ impl McpGlobalPolicyState {
             read_only: self.read_only,
             allow_dangerous_sql: self.allow_dangerous_sql,
             allowed_connection_ids: self.allowed_connection_ids.clone(),
+            allowed_group_ids: self.allowed_group_ids.clone(),
             allowed_tool_names: self.allowed_tool_names.clone(),
             connection_policies: self.connection_policies.clone(),
+            group_policies: self.group_policies.clone(),
             query_timeout_secs: self.query_timeout_secs,
         }
     }
@@ -400,6 +425,8 @@ impl McpGlobalPolicy {
             ids.dedup();
             ids
         });
+        let allowed_group_ids =
+            if allowed_connection_ids.is_some() { normalize_mcp_ids(&self.allowed_group_ids) } else { Vec::new() };
         let allowed_tool_names = self.allowed_tool_names.as_ref().map(|tools| {
             let mut tools = tools
                 .iter()
@@ -481,17 +508,52 @@ impl McpGlobalPolicy {
             }
         }
 
+        let mut group_policies = HashMap::<String, McpGroupPolicy>::new();
+        for rule in &self.group_policies {
+            let group_id = rule.group_id.trim();
+            if group_id.is_empty() {
+                continue;
+            }
+            group_policies
+                .entry(group_id.to_string())
+                .and_modify(|current| {
+                    current.read_only |= rule.read_only;
+                    current.allow_dangerous_sql &= rule.allow_dangerous_sql;
+                })
+                .or_insert_with(|| McpGroupPolicy {
+                    group_id: group_id.to_string(),
+                    read_only: rule.read_only,
+                    allow_dangerous_sql: !rule.read_only && rule.allow_dangerous_sql,
+                });
+        }
+        let mut group_policies = group_policies.into_values().collect::<Vec<_>>();
+        group_policies.sort_by(|left, right| left.group_id.cmp(&right.group_id));
+        for rule in &mut group_policies {
+            if rule.read_only {
+                rule.allow_dangerous_sql = false;
+            }
+        }
+
         Self {
             read_only: self.read_only,
             allow_dangerous_sql: !self.read_only && self.allow_dangerous_sql,
             allowed_connection_ids,
+            allowed_group_ids,
             allowed_tool_names,
             connection_policies,
+            group_policies,
             query_timeout_secs: self.query_timeout_secs,
         }
     }
 }
 
+fn normalize_mcp_ids(ids: &[String]) -> Vec<String> {
+    let mut ids = ids.iter().map(|id| id.trim()).filter(|id| !id.is_empty()).map(ToOwned::to_owned).collect::<Vec<_>>();
+    ids.sort();
+    ids.dedup();
+    ids
+}
+
 fn normalize_mcp_database_names(databases: &[String]) -> Vec<String> {
     let mut databases = databases
         .iter()
@@ -2034,8 +2096,10 @@ impl Storage {
                         read_only: policy.read_only,
                         allow_dangerous_sql: policy.allow_dangerous_sql,
                         allowed_connection_ids: policy.allowed_connection_ids,
+                        allowed_group_ids: policy.allowed_group_ids,
                         allowed_tool_names: policy.allowed_tool_names,
                         connection_policies: policy.connection_policies,
+                        group_policies: policy.group_policies,
                         query_timeout_secs: policy.query_timeout_secs,
                     });
                 };
@@ -2048,8 +2112,10 @@ impl Storage {
                         read_only: policy.read_only,
                         allow_dangerous_sql: policy.allow_dangerous_sql,
                         allowed_connection_ids: policy.allowed_connection_ids,
+                        allowed_group_ids: policy.allowed_group_ids,
                         allowed_tool_names: policy.allowed_tool_names,
                         connection_policies: policy.connection_policies,
+                        group_policies: policy.group_policies,
                         query_timeout_secs: policy.query_timeout_secs,
                     });
                 };
@@ -2061,8 +2127,10 @@ impl Storage {
                     read_only: policy.read_only,
                     allow_dangerous_sql: policy.allow_dangerous_sql,
                     allowed_connection_ids: policy.allowed_connection_ids,
+                    allowed_group_ids: policy.allowed_group_ids,
                     allowed_tool_names: policy.allowed_tool_names,
                     connection_policies: policy.connection_policies,
+                    group_policies: policy.group_policies,
                     query_timeout_secs: policy.query_timeout_secs,
                 })
             })
@@ -3052,7 +3120,24 @@ fn ensure_mcp_connection_change_allowed_in_tx(
         );
     }
     if let Some(connection_id) = target_connection_id {
-        if policy.allowed_connection_ids.as_ref().is_some_and(|ids| !ids.iter().any(|id| id == connection_id)) {
+        let group_paths = if crate::mcp_policy::policy_uses_connection_groups(&policy) {
+            let layout_json: Option<String> = tx
+                .query_row("SELECT layout_json FROM sidebar_layout WHERE id = 1", [], |row| row.get(0))
+                .optional()
+                .map_err(|error| format!("MCP_POLICY_UNAVAILABLE: {error}"))?;
+            layout_json
+                .map(|json| {
+                    let layout = serde_json::from_str(&json)
+                        .map_err(|error| format!("MCP_POLICY_UNAVAILABLE: invalid sidebar layout JSON: {error}"))?;
+                    crate::mcp_policy::connection_group_paths(&layout)
+                        .map_err(|error| format!("MCP_POLICY_UNAVAILABLE: {error}"))
+                })
+                .transpose()?
+                .unwrap_or_default()
+        } else {
+            HashMap::new()
+        };
+        if !crate::mcp_policy::policy_allows_connection(&policy, group_paths.get(connection_id), connection_id) {
             return Err(format!(
                 "CONNECTION_OUT_OF_SCOPE: connection '{connection_id}' is not allowed by the current DBX MCP policy"
             ));
@@ -6584,8 +6669,10 @@ mod tests {
                 read_only: false,
                 allow_dangerous_sql: false,
                 allowed_connection_ids: None,
+                allowed_group_ids: Vec::new(),
                 allowed_tool_names: None,
                 connection_policies: Vec::new(),
+                group_policies: Vec::new(),
                 query_timeout_secs: None,
             }
         );
@@ -6609,8 +6696,10 @@ mod tests {
                 read_only: true,
                 allow_dangerous_sql: false,
                 allowed_connection_ids: Some(vec!["conn-1".to_string(), "conn-2".to_string()]),
+                allowed_group_ids: Vec::new(),
                 allowed_tool_names: None,
                 connection_policies: Vec::new(),
+                group_policies: Vec::new(),
                 query_timeout_secs: Some(120),
             }
         );

crates/dbx-core/src/table_import.rs

@@ -10502,7 +10502,7 @@ mod tests {
         )
         .await;
 
-        assert!(!state.pool_handle(pool_key).await.is_some());
+        assert!(state.pool_handle(pool_key).await.is_none());
     }
 
     #[test]