fix: grant default CONNECT action for new created user (#7716)

To fix #7596 , since we don't have concept of PUBLIC group yet, for newly create user, this PR will simply grant CONNECT action of the current database in session. Note that, like PostgreSQL we still need to grant other actions for new user, so that he can do more operation rather than just connect.
```sql
~ psql -h localhost -p 4566 -d dev -U root
psql (14.5 (Homebrew), server 9.5.0)
SSL connection (protocol: TLSv1.3, cipher: TLS_AES_256_GCM_SHA384, bits: 256, compression: off)
Type "help" for help.

dev=> create table a (v1 integer);
CREATE_TABLE
dev=> create user xxx with password 'abc';
CREATE_USER
dev=> \q
~ psql -h localhost -p 4566 -d dev -U xxx
Password for user xxx:
psql (14.5 (Homebrew), server 9.5.0)
SSL connection (protocol: TLSv1.3, cipher: TLS_AES_256_GCM_SHA384, bits: 256, compression: off)
Type "help" for help.

dev=> select * from a;
ERROR:  QueryError: Permission denied: Do not have the privilege
dev=> \q
~ psql -h localhost -p 4566 -d dev -U root
psql (14.5 (Homebrew), server 9.5.0)
SSL connection (protocol: TLSv1.3, cipher: TLS_AES_256_GCM_SHA384, bits: 256, compression: off)
Type "help" for help.

dev=> create materialized view mv1 as select * from a;
CREATE_MATERIALIZED_VIEW
dev=> grant all on materialized view mv1 to xxx;
GRANT_PRIVILEGE
dev=> insert into a values (1),(2),(3);
INSERT 0 3
dev=> \q
~ psql -h localhost -p 4566 -d dev -U xxx
Password for user xxx:
psql (14.5 (Homebrew), server 9.5.0)
SSL connection (protocol: TLSv1.3, cipher: TLS_AES_256_GCM_SHA384, bits: 256, compression: off)
Type "help" for help.

dev=> select * from mv1;
v1
----
3
2
1
(3 rows)
```

Approved-By: chenzl25
Approved-By: xxchan

src/frontend/src/handler/create_user.rs

@@ -15,19 +15,21 @@
 use pgwire::pg_response::{PgResponse, StatementType};
 use risingwave_common::error::ErrorCode::PermissionDenied;
 use risingwave_common::error::Result;
-use risingwave_pb::user::UserInfo;
+use risingwave_pb::user::grant_privilege::{Action, ActionWithGrantOption, Object};
+use risingwave_pb::user::{GrantPrivilege, UserInfo};
 use risingwave_sqlparser::ast::{CreateUserStatement, UserOption, UserOptions};
 
 use super::RwPgResponse;
 use crate::binder::Binder;
-use crate::catalog::CatalogError;
+use crate::catalog::{CatalogError, DatabaseId};
 use crate::handler::HandlerArgs;
 use crate::user::user_authentication::encrypted_password;
 
 fn make_prost_user_info(
     user_name: String,
     options: &UserOptions,
     session_user: &UserInfo,
+    database_id: DatabaseId,
 ) -> Result<UserInfo> {
     if !session_user.is_super {
         let require_super = options
@@ -45,10 +47,22 @@ fn make_prost_user_info(
         }
     }
 
+    // Since we don't have concept of PUBLIC group yet, here we simply grant new user with CONNECT
+    // action of session database.
+    let grant_privileges = vec![GrantPrivilege {
+        action_with_opts: vec![ActionWithGrantOption {
+            action: Action::Connect as i32,
+            with_grant_option: true,
+            granted_by: session_user.id,
+        }],
+        object: Some(Object::DatabaseId(database_id)),
+    }];
+
     let mut user_info = UserInfo {
         name: user_name,
         // the LOGIN option is implied if it is not explicitly specified.
         can_login: true,
+        grant_privileges,
         ..Default::default()
     };
 
@@ -85,6 +99,13 @@ pub async fn handle_create_user(
     stmt: CreateUserStatement,
 ) -> Result<RwPgResponse> {
     let session = handler_args.session;
+    let database_id = {
+        let catalog_reader = session.env().catalog_reader().read_guard();
+        catalog_reader
+            .get_database_by_name(session.database())
+            .expect("session database should exist")
+            .id()
+    };
     let user_info = {
         let user_name = Binder::resolve_user_name(stmt.user_name)?;
         let user_reader = session.env().user_info_reader().read_guard();
@@ -96,7 +117,7 @@ pub async fn handle_create_user(
             .get_user_by_name(session.user_name())
             .ok_or_else(|| CatalogError::NotFound("user", session.user_name().to_string()))?;
 
-        make_prost_user_info(user_name, &stmt.with_options, session_user)?
+        make_prost_user_info(user_name, &stmt.with_options, session_user, database_id)?
     };
 
     let user_info_writer = session.env().user_info_writer();

src/frontend/src/handler/handle_privilege.rs

@@ -241,10 +241,16 @@ mod tests {
             .await
             .unwrap();
 
-        let database_id = {
+        let (session_database_id, database_id) = {
             let catalog_reader = session.env().catalog_reader();
             let reader = catalog_reader.read_guard();
-            reader.get_database_by_name("db1").unwrap().id()
+            (
+                reader
+                    .get_database_by_name(session.database())
+                    .unwrap()
+                    .id(),
+                reader.get_database_by_name("db1").unwrap().id(),
+            )
         };
 
         {
@@ -253,21 +259,31 @@ mod tests {
             let user_info = reader.get_user_by_name("user1").unwrap();
             assert_eq!(
                 user_info.grant_privileges,
-                vec![ProstPrivilege {
-                    action_with_opts: vec![
-                        ActionWithGrantOption {
+                vec![
+                    ProstPrivilege {
+                        action_with_opts: vec![ActionWithGrantOption {
                             action: Action::Connect as i32,
                             with_grant_option: true,
-                            granted_by: DEFAULT_SUPER_USER_ID,
-                        },
-                        ActionWithGrantOption {
-                            action: Action::Create as i32,
-                            with_grant_option: true,
-                            granted_by: DEFAULT_SUPER_USER_ID,
-                        }
-                    ],
-                    object: Some(ProstObject::DatabaseId(database_id)),
-                }]
+                            granted_by: session.user_id(),
+                        }],
+                        object: Some(ProstObject::DatabaseId(session_database_id)),
+                    },
+                    ProstPrivilege {
+                        action_with_opts: vec![
+                            ActionWithGrantOption {
+                                action: Action::Connect as i32,
+                                with_grant_option: true,
+                                granted_by: DEFAULT_SUPER_USER_ID,
+                            },
+                            ActionWithGrantOption {
+                                action: Action::Create as i32,
+                                with_grant_option: true,
+                                granted_by: DEFAULT_SUPER_USER_ID,
+                            }
+                        ],
+                        object: Some(ProstObject::DatabaseId(database_id)),
+                    }
+                ]
             );
         }
 
@@ -282,6 +298,7 @@ mod tests {
             assert!(user_info
                 .grant_privileges
                 .iter()
+                .filter(|gp| gp.object == Some(ProstObject::DatabaseId(database_id)))
                 .all(|p| p.action_with_opts.iter().all(|ao| !ao.with_grant_option)));
         }
 
@@ -293,7 +310,18 @@ mod tests {
             let user_reader = session.env().user_info_reader();
             let reader = user_reader.read_guard();
             let user_info = reader.get_user_by_name("user1").unwrap();
-            assert!(user_info.grant_privileges.is_empty());
+            assert_eq!(
+                user_info.grant_privileges,
+                vec![ProstPrivilege {
+                    action_with_opts: vec![ActionWithGrantOption {
+                        action: Action::Connect as i32,
+                        with_grant_option: true,
+                        granted_by: session.user_id(),
+                    }],
+                    object: Some(ProstObject::DatabaseId(session_database_id)),
+                }]
+            );
         }
+        frontend.run_sql("DROP USER user1").await.unwrap();
     }
 }

src/meta/src/manager/catalog/mod.rs

@@ -1614,12 +1614,6 @@ where
                 user.name
             )));
         }
-        if !user.grant_privileges.is_empty() {
-            return Err(MetaError::permission_denied(format!(
-                "Cannot drop user {} with privileges",
-                id
-            )));
-        }
         if user_core
             .user_grant_relation
             .get(&id)

src/meta/src/rpc/service/user_service.rs

@@ -27,8 +27,6 @@ use crate::manager::{CatalogManagerRef, IdCategory, MetaSrvEnv};
 use crate::storage::MetaStore;
 use crate::MetaResult;
 
-// TODO: Change user manager as a part of the catalog manager, to ensure that operations on Catalog
-// and User are transactional.
 pub struct UserServiceImpl<S: MetaStore> {
     env: MetaSrvEnv<S>,