feat: use bit flags for user privilege check (#13033)

proto/user.proto

@@ -43,6 +43,7 @@ message GrantPrivilege {
     DELETE = 4;
     CREATE = 5;
     CONNECT = 6;
+    USAGE = 7;
   }
 
   message ActionWithGrantOption {

src/common/Cargo.toml

@@ -38,6 +38,7 @@ easy-ext = "1"
 educe = "0.4"
 either = "1"
 enum-as-inner = "0.6"
+enumflags2 = { version = "0.7.8" }
 ethnum = { version = "1", features = ["serde"] }
 fixedbitset = { version = "0.4", features = ["std"] }
 fs-err = "2"

src/common/src/acl/mod.rs

@@ -0,0 +1,144 @@
+// Copyright 2023 RisingWave 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.
+
+//! `Acl` defines all grantable privileges.
+
+use std::convert::Into;
+use std::fmt::Formatter;
+use std::sync::LazyLock;
+
+use enumflags2::{bitflags, make_bitflags, BitFlags};
+use parse_display::Display;
+use risingwave_pb::user::grant_privilege::PbAction;
+
+#[bitflags]
+#[repr(u64)]
+#[derive(Clone, Copy, Debug, Display, Eq, PartialEq)]
+pub enum AclMode {
+    #[display("a")]
+    Insert = 1 << 0, // formerly known as "append".
+    #[display("r")]
+    Select = 1 << 1, // formerly known as "read".
+    #[display("w")]
+    Update = 1 << 2, // formerly known as "write".
+    #[display("d")]
+    Delete = 1 << 3,
+    #[display("D")]
+    Truncate = 1 << 4, // super-delete, as it were
+    #[display("x")]
+    References = 1 << 5,
+    #[display("t")]
+    Trigger = 1 << 6,
+    #[display("X")]
+    Execute = 1 << 7, // For functions
+    #[display("U")]
+    Usage = 1 << 8, // For various object types
+    #[display("C")]
+    Create = 1 << 9, // For namespaces and databases
+    #[display("T")]
+    CreateTemp = 1 << 10, // For databases
+    #[display("c")]
+    Connect = 1 << 11, // For databases
+    #[display("s")]
+    Set = 1 << 12, // For configuration parameters
+    #[display("A")]
+    AlterSystem = 1 << 13, // For configuration parameters
+    #[display("m")]
+    Maintain = 1 << 14, // For relations
+}
+
+impl From<PbAction> for AclMode {
+    fn from(action: PbAction) -> Self {
+        match action {
+            PbAction::Unspecified => unreachable!(),
+            PbAction::Select => AclMode::Select,
+            PbAction::Insert => AclMode::Insert,
+            PbAction::Update => AclMode::Update,
+            PbAction::Delete => AclMode::Delete,
+            PbAction::Create => AclMode::Create,
+            PbAction::Connect => AclMode::Connect,
+            PbAction::Usage => AclMode::Usage,
+        }
+    }
+}
+
+impl From<AclMode> for PbAction {
+    fn from(val: AclMode) -> Self {
+        match val {
+            AclMode::Select => PbAction::Select,
+            AclMode::Insert => PbAction::Insert,
+            AclMode::Update => PbAction::Update,
+            AclMode::Delete => PbAction::Delete,
+            AclMode::Create => PbAction::Create,
+            AclMode::Connect => PbAction::Connect,
+            AclMode::Usage => PbAction::Usage,
+            _ => unreachable!(),
+        }
+    }
+}
+
+/// `AclModeSet` defines a set of `AclMode`s.
+#[derive(Clone, Debug)]
+pub struct AclModeSet {
+    pub modes: BitFlags<AclMode>,
+}
+
+pub static ALL_AVAILABLE_DATABASE_MODES: LazyLock<AclModeSet> =
+    LazyLock::new(|| make_bitflags!(AclMode::{Create | Connect}).into());
+pub static ALL_AVAILABLE_SCHEMA_MODES: LazyLock<AclModeSet> =
+    LazyLock::new(|| make_bitflags!(AclMode::{Create | Usage}).into());
+pub static ALL_AVAILABLE_TABLE_MODES: LazyLock<AclModeSet> =
+    LazyLock::new(|| make_bitflags!(AclMode::{Select | Insert | Update | Delete}).into());
+pub static ALL_AVAILABLE_SOURCE_MODES: LazyLock<AclModeSet> = LazyLock::new(AclModeSet::readonly);
+pub static ALL_AVAILABLE_MVIEW_MODES: LazyLock<AclModeSet> = LazyLock::new(AclModeSet::readonly);
+pub static ALL_AVAILABLE_VIEW_MODES: LazyLock<AclModeSet> = LazyLock::new(AclModeSet::readonly);
+pub static ALL_AVAILABLE_SINK_MODES: LazyLock<AclModeSet> = LazyLock::new(AclModeSet::empty);
+pub static ALL_AVAILABLE_FUNCTION_MODES: LazyLock<AclModeSet> =
+    LazyLock::new(|| BitFlags::from(AclMode::Execute).into());
+pub static ALL_AVAILABLE_CONNECTION_MODES: LazyLock<AclModeSet> =
+    LazyLock::new(|| BitFlags::from(AclMode::Usage).into());
+
+impl AclModeSet {
+    pub fn empty() -> Self {
+        Self {
+            modes: BitFlags::empty(),
+        }
+    }
+
+    pub fn readonly() -> Self {
+        Self {
+            modes: BitFlags::from(AclMode::Select),
+        }
+    }
+
+    pub fn has_mode(&self, mode: AclMode) -> bool {
+        self.modes.contains(mode)
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = AclMode> + '_ {
+        self.modes.iter()
+    }
+}
+
+impl From<BitFlags<AclMode>> for AclModeSet {
+    fn from(modes: BitFlags<AclMode>) -> Self {
+        Self { modes }
+    }
+}
+
+impl std::fmt::Display for AclModeSet {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.modes)
+    }
+}

src/common/src/lib.rs

@@ -70,6 +70,7 @@ pub mod system_param;
 pub mod telemetry;
 pub mod transaction;
 
+pub mod acl;
 pub mod metrics;
 pub mod test_utils;
 pub mod types;

src/frontend/src/catalog/system_catalog/mod.rs

@@ -21,15 +21,15 @@ use std::sync::{Arc, LazyLock};
 
 use async_trait::async_trait;
 use itertools::Itertools;
+use risingwave_common::acl::AclMode;
 use risingwave_common::catalog::{
     ColumnCatalog, ColumnDesc, Field, SysCatalogReader, TableDesc, TableId, DEFAULT_SUPER_USER_ID,
     NON_RESERVED_SYS_CATALOG_ID,
 };
 use risingwave_common::error::Result;
 use risingwave_common::row::OwnedRow;
 use risingwave_common::types::DataType;
-use risingwave_pb::user::grant_privilege::{Action, Object};
-use risingwave_pb::user::UserInfo;
+use risingwave_pb::user::grant_privilege::Object;
 
 use crate::catalog::catalog_service::CatalogReader;
 use crate::catalog::system_catalog::information_schema::*;
@@ -39,6 +39,7 @@ use crate::catalog::view_catalog::ViewCatalog;
 use crate::meta_client::FrontendMetaClient;
 use crate::scheduler::worker_node_manager::WorkerNodeManagerRef;
 use crate::session::AuthContext;
+use crate::user::user_catalog::UserCatalog;
 use crate::user::user_privilege::available_prost_privilege;
 use crate::user::user_service::UserInfoReader;
 use crate::user::UserId;
@@ -212,17 +213,17 @@ fn infer_dummy_view_sql(columns: &[SystemCatalogColumnsDef<'_>]) -> String {
 fn get_acl_items(
     object: &Object,
     for_dml_table: bool,
-    users: &Vec<UserInfo>,
+    users: &Vec<UserCatalog>,
     username_map: &HashMap<UserId, String>,
 ) -> String {
     let mut res = String::from("{");
     let mut empty_flag = true;
     let super_privilege = available_prost_privilege(object.clone(), for_dml_table);
     for user in users {
-        let privileges = if user.get_is_super() {
+        let privileges = if user.is_super {
             vec![&super_privilege]
         } else {
-            user.get_grant_privileges()
+            user.grant_privileges
                 .iter()
                 .filter(|&privilege| privilege.object.as_ref().unwrap() == object)
                 .collect_vec()
@@ -233,43 +234,29 @@ fn get_acl_items(
         let mut grantor_map = HashMap::new();
         privileges.iter().for_each(|&privilege| {
             privilege.action_with_opts.iter().for_each(|ao| {
-                grantor_map.entry(ao.granted_by).or_insert_with(Vec::new);
                 grantor_map
-                    .get_mut(&ao.granted_by)
-                    .unwrap()
-                    .push((ao.action, ao.with_grant_option));
+                    .entry(ao.granted_by)
+                    .or_insert_with(Vec::new)
+                    .push((ao.get_action().unwrap(), ao.with_grant_option));
             })
         });
-        for key in grantor_map.keys() {
+        for (granted_by, actions) in grantor_map {
             if empty_flag {
                 empty_flag = false;
             } else {
                 res.push(',');
             }
-            res.push_str(user.get_name());
+            res.push_str(&user.name);
             res.push('=');
-            grantor_map
-                .get(key)
-                .unwrap()
-                .iter()
-                .for_each(|(action, option)| {
-                    let str = match Action::try_from(*action).unwrap() {
-                        Action::Select => "r",
-                        Action::Insert => "a",
-                        Action::Update => "w",
-                        Action::Delete => "d",
-                        Action::Create => "C",
-                        Action::Connect => "c",
-                        _ => unreachable!(),
-                    };
-                    res.push_str(str);
-                    if *option {
-                        res.push('*');
-                    }
-                });
+            for (action, option) in actions {
+                res.push_str(&AclMode::from(action).to_string());
+                if option {
+                    res.push('*');
+                }
+            }
             res.push('/');
             // should be able to query grantor's name
-            res.push_str(username_map.get(key).as_ref().unwrap());
+            res.push_str(username_map.get(&granted_by).unwrap());
         }
     }
     res.push('}');

src/frontend/src/handler/alter_user.rs

@@ -24,11 +24,12 @@ use crate::binder::Binder;
 use crate::catalog::CatalogError;
 use crate::handler::HandlerArgs;
 use crate::user::user_authentication::encrypted_password;
+use crate::user::user_catalog::UserCatalog;
 
 fn alter_prost_user_info(
     mut user_info: UserInfo,
     options: &UserOptions,
-    session_user: &UserInfo,
+    session_user: &UserCatalog,
 ) -> Result<(UserInfo, Vec<UpdateField>)> {
     if !session_user.is_super {
         let require_super = user_info.is_super
@@ -116,7 +117,7 @@ fn alter_prost_user_info(
 fn alter_rename_prost_user_info(
     mut user_info: UserInfo,
     new_name: ObjectName,
-    session_user: &UserInfo,
+    session_user: &UserCatalog,
 ) -> Result<(UserInfo, Vec<UpdateField>)> {
     if session_user.id == user_info.id {
         return Err(InternalError("session user cannot be renamed".to_string()).into());
@@ -153,7 +154,7 @@ pub async fn handle_alter_user(
         let old_info = user_reader
             .get_user_by_name(&user_name)
             .ok_or(CatalogError::NotFound("user", user_name))?
-            .clone();
+            .to_prost();
 
         let session_user = user_reader
             .get_user_by_name(session.user_name())

src/frontend/src/handler/create_index.rs

@@ -18,12 +18,13 @@ use std::rc::Rc;
 use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use pgwire::pg_response::{PgResponse, StatementType};
+use risingwave_common::acl::AclMode;
 use risingwave_common::catalog::{IndexId, TableDesc, TableId};
 use risingwave_common::error::{ErrorCode, Result};
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 use risingwave_pb::catalog::{PbIndex, PbStreamJobStatus, PbTable};
 use risingwave_pb::stream_plan::stream_fragment_graph::Parallelism;
-use risingwave_pb::user::grant_privilege::{Action, Object};
+use risingwave_pb::user::grant_privilege::Object;
 use risingwave_sqlparser::ast;
 use risingwave_sqlparser::ast::{Ident, ObjectName, OrderByExpr};
 
@@ -74,7 +75,7 @@ pub(crate) fn gen_create_index_plan(
 
     session.check_privileges(&[ObjectCheckItem::new(
         table.owner,
-        Action::Select,
+        AclMode::Select,
         Object::TableId(table.id.table_id),
     )])?;
 

src/frontend/src/handler/create_mv.rs

@@ -14,10 +14,10 @@
 
 use itertools::Itertools;
 use pgwire::pg_response::{PgResponse, StatementType};
+use risingwave_common::acl::AclMode;
 use risingwave_common::error::{ErrorCode, Result};
 use risingwave_pb::catalog::{CreateType, PbTable};
 use risingwave_pb::stream_plan::stream_fragment_graph::Parallelism;
-use risingwave_pb::user::grant_privilege::Action;
 use risingwave_sqlparser::ast::{EmitMode, Ident, ObjectName, Query};
 
 use super::privilege::resolve_relation_privileges;
@@ -66,7 +66,7 @@ pub(super) fn get_column_names(
         }
         if let Some(relation) = &select.from {
             let mut check_items = Vec::new();
-            resolve_relation_privileges(relation, Action::Select, &mut check_items);
+            resolve_relation_privileges(relation, AclMode::Select, &mut check_items);
             session.check_privileges(&check_items)?;
         }
     }