feat(rbac): RBAC support connection object

1.   **New Configuration Parameter:** Introduces `enable_experimental_connection_rbac_check` to toggle RBAC permission verification for connections.   Disabled by default for backward compatibility.
2.   **Global Privileges:** Adds `CREATE CONNECTION` and `ACCESS CONNECTION` global privileges governing connection creation and unrestricted usage rights respectively.
3.   **Ownership Model:** Implements `OWNERSHIP` semantics, allowing privileged users/roles to perform arbitrary DDL operations on connections.
4.   **Show grants on connection <connection_name>.

Author: TCeason

src/meta/app/src/principal/ownership_object.rs

@@ -56,6 +56,10 @@ pub enum OwnershipObject {
     Warehouse {
         id: String,
     },
+
+    Connection {
+        name: String,
+    },
 }
 
 impl OwnershipObject {
@@ -84,7 +88,8 @@ impl fmt::Display for OwnershipObject {
             }
             OwnershipObject::UDF { name } => write!(f, "UDF {name}"),
             OwnershipObject::Stage { name } => write!(f, "STAGE {name}"),
-            OwnershipObject::Warehouse { id } => write!(f, "Warehouse {id}"),
+            OwnershipObject::Warehouse { id } => write!(f, "WAREHOUSE {id}"),
+            OwnershipObject::Connection { name } => write!(f, "CONNECTION {name}"),
         }
     }
 }
@@ -125,6 +130,7 @@ impl KeyCodec for OwnershipObject {
             OwnershipObject::Stage { name } => b.push_raw("stage-by-name").push_str(name),
             OwnershipObject::UDF { name } => b.push_raw("udf-by-name").push_str(name),
             OwnershipObject::Warehouse { id } => b.push_raw("warehouse-by-id").push_str(id),
+            OwnershipObject::Connection { name } => b.push_raw("connection-by-name").push_str(name),
         }
     }
 
@@ -175,9 +181,13 @@ impl KeyCodec for OwnershipObject {
                 let id = p.next_str()?;
                 Ok(OwnershipObject::Warehouse { id })
             }
+            "connection-by-name" => {
+                let name = p.next_str()?;
+                Ok(OwnershipObject::Connection { name })
+            }
             _ => Err(kvapi::KeyError::InvalidSegment {
                 i: p.index(),
-                expect: "database-by-id|database-by-catalog-id|table-by-id|table-by-catalog-id|stage-by-name|udf-by-name|warehouse-by-id"
+                expect: "database-by-id|database-by-catalog-id|table-by-id|table-by-catalog-id|stage-by-name|udf-by-name|warehouse-by-id|connection-by-name"
                     .to_string(),
                 got: q.to_string(),
             }),

src/meta/app/src/principal/user_grant.rs

@@ -60,6 +60,7 @@ pub enum GrantObject {
     UDF(String),
     Stage(String),
     Warehouse(String),
+    Connection(String),
 }
 
 impl GrantObject {
@@ -93,6 +94,7 @@ impl GrantObject {
             (GrantObject::Stage(lstage), GrantObject::Stage(rstage)) => lstage == rstage,
             (GrantObject::UDF(udf), GrantObject::UDF(rudf)) => udf == rudf,
             (GrantObject::Warehouse(w), GrantObject::Warehouse(rw)) => w == rw,
+            (GrantObject::Connection(c), GrantObject::Connection(rc)) => c == rc,
             _ => false,
         }
     }
@@ -116,6 +118,9 @@ impl GrantObject {
             GrantObject::Warehouse(_) => {
                 UserPrivilegeSet::available_privileges_on_warehouse(available_ownership)
             }
+            GrantObject::Connection(_) => {
+                UserPrivilegeSet::available_privileges_on_connection(available_ownership)
+            }
         }
     }
 
@@ -124,7 +129,8 @@ impl GrantObject {
             GrantObject::Global
             | GrantObject::Stage(_)
             | GrantObject::UDF(_)
-            | GrantObject::Warehouse(_) => None,
+            | GrantObject::Warehouse(_)
+            | GrantObject::Connection(_) => None,
             GrantObject::Database(cat, _) | GrantObject::DatabaseById(cat, _) => Some(cat.clone()),
             GrantObject::Table(cat, _, _) | GrantObject::TableById(cat, _, _) => Some(cat.clone()),
         }
@@ -146,6 +152,7 @@ impl fmt::Display for GrantObject {
             GrantObject::UDF(udf) => write!(f, "UDF {udf}"),
             GrantObject::Stage(stage) => write!(f, "STAGE {stage}"),
             GrantObject::Warehouse(w) => write!(f, "WAREHOUSE {w}"),
+            GrantObject::Connection(c) => write!(f, "CONNECTION {c}"),
         }
     }
 }

src/meta/app/src/principal/user_privilege.rs

@@ -78,6 +78,10 @@ pub enum UserPrivilegeType {
     CreateDatabase = 1 << 20,
     // Privilege to Create warehouse
     CreateWarehouse = 1 << 21,
+    // Privilege to Create Connection
+    CreateConnection = 1 << 22,
+    // Privilege to Access Connection
+    AccessConnection = 1 << 23,
     // Discard Privilege Type
     Set = 1 << 4,
 }
@@ -105,6 +109,8 @@ const ALL_PRIVILEGES: BitFlags<UserPrivilegeType> = make_bitflags!(
         | Write
         | CreateDatabase
         | CreateWarehouse
+        | CreateConnection
+        | AccessConnection
     }
 );
 
@@ -133,6 +139,8 @@ impl Display for UserPrivilegeType {
             UserPrivilegeType::Write => "Write",
             UserPrivilegeType::CreateDatabase => "CREATE DATABASE",
             UserPrivilegeType::CreateWarehouse => "CREATE WAREHOUSE",
+            UserPrivilegeType::CreateConnection => "CREATE CONNECTION",
+            UserPrivilegeType::AccessConnection => "ACCESS CONNECTION",
         })
     }
 }
@@ -173,6 +181,12 @@ impl From<databend_common_ast::ast::UserPrivilegeType> for UserPrivilegeType {
             databend_common_ast::ast::UserPrivilegeType::CreateWarehouse => {
                 UserPrivilegeType::CreateWarehouse
             }
+            databend_common_ast::ast::UserPrivilegeType::CreateConnection => {
+                UserPrivilegeType::CreateConnection
+            }
+            databend_common_ast::ast::UserPrivilegeType::AccessConnection => {
+                UserPrivilegeType::AccessConnection
+            }
             databend_common_ast::ast::UserPrivilegeType::Set => UserPrivilegeType::Set,
         }
     }
@@ -207,11 +221,13 @@ impl UserPrivilegeSet {
         let stage_privs_without_ownership = Self::available_privileges_on_stage(false);
         let udf_privs_without_ownership = Self::available_privileges_on_udf(false);
         let wh_privs_without_ownership = Self::available_privileges_on_warehouse(false);
-        let privs = make_bitflags!(UserPrivilegeType::{ Usage | Super | CreateUser | DropUser | CreateRole | DropRole | CreateDatabase | Grant | CreateDataMask | CreateWarehouse });
+        let connection_privs_without_ownership = Self::available_privileges_on_connection(false);
+        let privs = make_bitflags!(UserPrivilegeType::{ Usage | Super | CreateUser | DropUser | CreateRole | DropRole | CreateDatabase | Grant | CreateDataMask | CreateWarehouse | CreateConnection });
         (database_privs.privileges
             | privs
             | stage_privs_without_ownership.privileges
             | wh_privs_without_ownership.privileges
+            | connection_privs_without_ownership.privileges
             | udf_privs_without_ownership.privileges)
             .into()
     }
@@ -251,6 +267,14 @@ impl UserPrivilegeSet {
         }
     }
 
+    pub fn available_privileges_on_connection(available_ownership: bool) -> Self {
+        if available_ownership {
+            make_bitflags!(UserPrivilegeType::{ AccessConnection | Ownership }).into()
+        } else {
+            make_bitflags!(UserPrivilegeType::{ AccessConnection }).into()
+        }
+    }
+
     pub fn available_privileges_on_udf(available_ownership: bool) -> Self {
         if available_ownership {
             make_bitflags!(UserPrivilegeType::{ Usage | Ownership }).into()

src/meta/proto-conv/src/ownership_from_to_protobuf_impl.rs

@@ -91,6 +91,9 @@ impl FromToProto for mt::principal::OwnershipObject {
             pb::ownership_object::Object::Warehouse(
                 pb::ownership_object::OwnershipWarehouseObject { id },
             ) => Ok(mt::principal::OwnershipObject::Warehouse { id }),
+            pb::ownership_object::Object::Connection(
+                pb::ownership_object::OwnershipConnectionObject { connection },
+            ) => Ok(mt::principal::OwnershipObject::Connection { name: connection }),
         }
     }
 
@@ -131,6 +134,13 @@ impl FromToProto for mt::principal::OwnershipObject {
                     pb::ownership_object::OwnershipWarehouseObject { id: id.clone() },
                 ))
             }
+            mt::principal::OwnershipObject::Connection { name } => {
+                Some(pb::ownership_object::Object::Connection(
+                    pb::ownership_object::OwnershipConnectionObject {
+                        connection: name.clone(),
+                    },
+                ))
+            }
         };
         Ok(pb::OwnershipObject {
             ver: VER,

src/meta/proto-conv/src/user_from_to_protobuf_impl.rs

@@ -194,6 +194,9 @@ impl FromToProto for mt::principal::GrantObject {
             pb::grant_object::Object::Warehouse(pb::grant_object::GrantWarehouseObject {
                 warehouse,
             }) => Ok(mt::principal::GrantObject::Warehouse(warehouse)),
+            pb::grant_object::Object::Connection(pb::grant_object::GrantConnectionObject {
+                connection,
+            }) => Ok(mt::principal::GrantObject::Connection(connection)),
         }
     }
 
@@ -241,6 +244,11 @@ impl FromToProto for mt::principal::GrantObject {
                     warehouse: w.clone(),
                 },
             )),
+            mt::principal::GrantObject::Connection(c) => Some(
+                pb::grant_object::Object::Connection(pb::grant_object::GrantConnectionObject {
+                    connection: c.clone(),
+                }),
+            ),
         };
         Ok(pb::GrantObject {
             ver: VER,

src/meta/proto-conv/src/util.rs

@@ -165,6 +165,7 @@ const META_CHANGE_LOG: &[(u64, &str)] = &[
     (133, "2025-06-25: Add: Add new StageFileCompression Zip"),
     (134, "2025-06-27: Add: SequenceMeta.storage_version"),
     (135, "2025-07-16: Add: UDFServer.immutable, UDFScript.immutable"),
+    (136, "2025-07-18: Add: GrantConnectionObject and UserPrivilegeType AccessConnection, AccessConnection"),
     // Dear developer:
     //      If you're gonna add a new metadata version, you'll have to add a test for it.
     //      You could just copy an existing test file(e.g., `../tests/it/v024_table_meta.rs`)

src/meta/proto-conv/tests/it/main.rs

@@ -127,3 +127,4 @@ mod v132_remove_sequence_meta_start;
 mod v133_stage_file_compression;
 mod v134_add_sequence_meta_storage_version;
 mod v135_udf_immutable;
+mod v136_add_grant_object_connection;

src/meta/proto-conv/tests/it/v136_add_grant_object_connection.rs

@@ -0,0 +1,122 @@
+// Copyright 2023 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::HashSet;
+
+use chrono::DateTime;
+use chrono::Utc;
+use databend_common_meta_app as mt;
+use databend_common_meta_app::principal::OwnershipObject;
+use databend_common_meta_app::principal::UserGrantSet;
+use databend_common_meta_app::principal::UserPrivilegeType;
+use enumflags2::make_bitflags;
+use fastrace::func_name;
+
+use crate::common;
+
+// These bytes are built when a new version in introduced,
+// and are kept for backward compatibility test.
+//
+// *************************************************************
+// * These messages should never be updated,                   *
+// * only be added when a new version is added,                *
+// * or be removed when an old version is no longer supported. *
+// *************************************************************
+//
+
+#[test]
+fn test_decode_v136_grant_object() -> anyhow::Result<()> {
+    let role_info_v136 = vec![
+        10, 2, 114, 49, 18, 214, 1, 10, 23, 10, 9, 10, 0, 160, 6, 136, 1, 168, 6, 24, 16, 128, 128,
+        128, 2, 160, 6, 136, 1, 168, 6, 24, 10, 27, 10, 13, 74, 4, 10, 2, 99, 49, 160, 6, 136, 1,
+        168, 6, 24, 16, 128, 128, 128, 4, 160, 6, 136, 1, 168, 6, 24, 10, 33, 10, 22, 18, 13, 10,
+        7, 100, 101, 102, 97, 117, 108, 116, 18, 2, 100, 98, 160, 6, 136, 1, 168, 6, 24, 16, 2,
+        160, 6, 136, 1, 168, 6, 24, 10, 37, 10, 26, 26, 17, 10, 7, 100, 101, 102, 97, 117, 108,
+        116, 18, 2, 100, 98, 26, 2, 116, 98, 160, 6, 136, 1, 168, 6, 24, 16, 2, 160, 6, 136, 1,
+        168, 6, 24, 10, 24, 10, 13, 34, 4, 10, 2, 102, 49, 160, 6, 136, 1, 168, 6, 24, 16, 1, 160,
+        6, 136, 1, 168, 6, 24, 10, 26, 10, 13, 42, 4, 10, 2, 115, 49, 160, 6, 136, 1, 168, 6, 24,
+        16, 128, 128, 32, 160, 6, 136, 1, 168, 6, 24, 10, 23, 10, 9, 10, 0, 160, 6, 136, 1, 168, 6,
+        24, 16, 254, 255, 191, 7, 160, 6, 136, 1, 168, 6, 24, 160, 6, 136, 1, 168, 6, 24, 26, 23,
+        49, 57, 55, 48, 45, 48, 49, 45, 48, 49, 32, 48, 48, 58, 48, 48, 58, 48, 48, 32, 85, 84, 67,
+        34, 23, 49, 57, 55, 48, 45, 48, 49, 45, 48, 49, 32, 48, 48, 58, 48, 48, 58, 48, 48, 32, 85,
+        84, 67, 160, 6, 136, 1, 168, 6, 24,
+    ];
+    let want = || mt::principal::RoleInfo {
+        name: "r1".to_string(),
+        grants: UserGrantSet::new(
+            vec![
+                mt::principal::GrantEntry::new(
+                    mt::principal::GrantObject::Global,
+                    make_bitflags!(UserPrivilegeType::{CreateConnection}),
+                ),
+                mt::principal::GrantEntry::new(
+                    mt::principal::GrantObject::Connection("c1".to_string()),
+                    make_bitflags!(UserPrivilegeType::{AccessConnection}),
+                ),
+                mt::principal::GrantEntry::new(
+                    mt::principal::GrantObject::Database("default".to_string(), "db".to_string()),
+                    make_bitflags!(UserPrivilegeType::{Create}),
+                ),
+                mt::principal::GrantEntry::new(
+                    mt::principal::GrantObject::Table(
+                        "default".to_string(),
+                        "db".to_string(),
+                        "tb".to_string(),
+                    ),
+                    make_bitflags!(UserPrivilegeType::{Create}),
+                ),
+                mt::principal::GrantEntry::new(
+                    mt::principal::GrantObject::UDF("f1".to_string()),
+                    make_bitflags!(UserPrivilegeType::{Usage}),
+                ),
+                mt::principal::GrantEntry::new(
+                    mt::principal::GrantObject::Stage("s1".to_string()),
+                    make_bitflags!(UserPrivilegeType::{Write}),
+                ),
+                // test new global privilege CreateConneciton, AccessConnection
+                mt::principal::GrantEntry::new(
+                    mt::principal::GrantObject::Global,
+                    make_bitflags!(UserPrivilegeType::{Create | Select | Insert | Update | Delete | Drop | Alter | Super | CreateUser | DropUser | CreateRole | DropRole | Grant | CreateStage | Set | CreateDataMask | Ownership | Read | Write | CreateWarehouse | CreateConnection | AccessConnection }),
+                ),
+            ],
+            HashSet::new(),
+        ),
+        created_on: DateTime::<Utc>::default(),
+        update_on: DateTime::<Utc>::default(),
+    };
+
+    common::test_pb_from_to(func_name!(), want())?;
+    common::test_load_old(func_name!(), role_info_v136.as_slice(), 136, want())?;
+
+    Ok(())
+}
+
+#[test]
+fn test_decode_v136_ownership() -> anyhow::Result<()> {
+    let ownership_info_v136 = vec![
+        10, 2, 114, 49, 18, 13, 50, 4, 10, 2, 99, 49, 160, 6, 136, 1, 168, 6, 24, 160, 6, 136, 1,
+        168, 6, 24,
+    ];
+
+    let want = || mt::principal::OwnershipInfo {
+        role: "r1".to_string(),
+        object: OwnershipObject::Connection {
+            name: "c1".to_string(),
+        },
+    };
+    common::test_pb_from_to(func_name!(), want())?;
+    common::test_load_old(func_name!(), ownership_info_v136.as_slice(), 136, want())?;
+
+    Ok(())
+}

src/meta/protos/proto/ownership.proto

@@ -50,11 +50,16 @@ message OwnershipObject {
     string id = 1;
   }
 
+    message OwnershipConnectionObject {
+      string connection = 1;
+    }
+
   oneof object {
     OwnershipDatabaseObject database = 1;
     OwnershipTableObject table = 2;
     OwnershipUdfObject udf = 3;
     OwnershipStageObject stage = 4;
     OwnershipWarehouseObject warehouse = 5;
+    OwnershipConnectionObject connection = 6;
   }
 }

src/meta/protos/proto/user.proto

@@ -81,6 +81,10 @@ message GrantObject {
     string warehouse = 1;
   }
 
+  message GrantConnectionObject {
+      string connection = 1;
+    }
+
   oneof object {
     GrantGlobalObject global = 1;
     GrantDatabaseObject database = 2;
@@ -90,6 +94,7 @@ message GrantObject {
     GrantDatabaseIdObject databasebyid = 6;
     GrantTableIdObject tablebyid = 7;
     GrantWarehouseObject warehouse = 8;
+    GrantConnectionObject connection = 9;
   }
 }