risingwave_meta/controller/
user.rs

1// Copyright 2023 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::{HashMap, HashSet};
16
17use itertools::Itertools;
18use risingwave_common::catalog::{DEFAULT_SUPER_USER, DEFAULT_SUPER_USER_FOR_PG};
19use risingwave_meta_model::object::ObjectType;
20use risingwave_meta_model::prelude::{Object, User, UserDefaultPrivilege, UserPrivilege};
21use risingwave_meta_model::user_privilege::Action;
22use risingwave_meta_model::{
23    AuthInfo, DatabaseId, DefaultPrivilegeId, PrivilegeId, SchemaId, UserId, object, user,
24    user_default_privilege, user_privilege,
25};
26use risingwave_pb::common::PbObjectType;
27use risingwave_pb::meta::subscribe_response::{
28    Info as NotificationInfo, Operation as NotificationOperation,
29};
30use risingwave_pb::user::update_user_request::PbUpdateField;
31use risingwave_pb::user::{PbAction, PbGrantPrivilege, PbUserInfo};
32use sea_orm::ActiveValue::Set;
33use sea_orm::sea_query::{OnConflict, SimpleExpr, Value};
34use sea_orm::{
35    ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, PaginatorTrait, QueryFilter,
36    QuerySelect, TransactionTrait,
37};
38
39use crate::controller::catalog::CatalogController;
40use crate::controller::utils::{
41    PartialUserPrivilege, check_user_name_duplicate, ensure_object_id,
42    ensure_privileges_not_referred, ensure_user_id, extract_grant_obj_id,
43    get_iceberg_related_object_ids, get_index_state_tables_by_table_id, get_internal_tables_by_id,
44    get_object_owner, get_referring_privileges_cascade, get_user_privilege, list_user_info_by_ids,
45    upsert_user_privileges,
46};
47use crate::manager::{FrontendNotificationBatch, NotificationVersion};
48use crate::{MetaError, MetaResult};
49
50impl CatalogController {
51    pub(crate) fn add_users_update_notifications(
52        notifications: &mut FrontendNotificationBatch<'_>,
53        user_infos: Vec<PbUserInfo>,
54    ) {
55        for info in user_infos {
56            notifications.add(NotificationOperation::Update, NotificationInfo::User(info));
57        }
58    }
59
60    pub(crate) async fn notify_users_update(
61        &self,
62        user_infos: Vec<PbUserInfo>,
63    ) -> Option<NotificationVersion> {
64        let mut notifications = self.frontend_notification_batch();
65        Self::add_users_update_notifications(&mut notifications, user_infos);
66        notifications.publish().await
67    }
68
69    pub async fn create_user(&self, pb_user: PbUserInfo) -> MetaResult<NotificationVersion> {
70        let inner = self.inner.write().await;
71        let txn = inner.db.begin().await?;
72        check_user_name_duplicate(&pb_user.name, &txn).await?;
73
74        let grant_privileges = pb_user.grant_privileges.clone();
75        let user: user::ActiveModel = pb_user.into();
76        let user = user.insert(&txn).await?;
77
78        if !grant_privileges.is_empty() {
79            let mut privileges = vec![];
80            for gp in &grant_privileges {
81                let id = extract_grant_obj_id(gp.get_object()?);
82                for action_with_opt in &gp.action_with_opts {
83                    privileges.push(user_privilege::ActiveModel {
84                        user_id: Set(user.user_id),
85                        oid: Set(id),
86                        granted_by: Set(action_with_opt.granted_by as _),
87                        action: Set(action_with_opt.get_action()?.into()),
88                        with_grant_option: Set(action_with_opt.with_grant_option),
89                        ..Default::default()
90                    });
91                }
92            }
93            UserPrivilege::insert_many(privileges).exec(&txn).await?;
94        }
95        txn.commit().await?;
96
97        let mut user_info: PbUserInfo = user.into();
98        user_info.grant_privileges = grant_privileges;
99        let version = self
100            .notify_frontend(
101                NotificationOperation::Add,
102                NotificationInfo::User(user_info),
103            )
104            .await;
105
106        Ok(version)
107    }
108
109    pub async fn update_user(
110        &self,
111        update_user: PbUserInfo,
112        update_fields: &[PbUpdateField],
113    ) -> MetaResult<NotificationVersion> {
114        let inner = self.inner.write().await;
115        let txn = inner.db.begin().await?;
116        let rename_flag = update_fields.contains(&PbUpdateField::Rename);
117        if rename_flag {
118            check_user_name_duplicate(&update_user.name, &txn).await?;
119        }
120
121        let user = User::find_by_id(update_user.id as UserId)
122            .one(&txn)
123            .await?
124            .ok_or_else(|| MetaError::catalog_id_not_found("user", update_user.id))?;
125        let mut user = user.into_active_model();
126        update_fields.iter().for_each(|&field| match field {
127            PbUpdateField::Unspecified => unreachable!(),
128            PbUpdateField::Super => user.is_super = Set(update_user.is_super),
129            PbUpdateField::Login => user.can_login = Set(update_user.can_login),
130            PbUpdateField::CreateDb => user.can_create_db = Set(update_user.can_create_db),
131            PbUpdateField::CreateUser => user.can_create_user = Set(update_user.can_create_user),
132            PbUpdateField::AuthInfo => {
133                user.auth_info = Set(update_user.auth_info.as_ref().map(AuthInfo::from))
134            }
135            PbUpdateField::Rename => user.name = Set(update_user.name.clone()),
136            PbUpdateField::Admin => user.is_admin = Set(update_user.is_admin),
137        });
138
139        let user = user.update(&txn).await?;
140        let mut user_info: PbUserInfo = user.into();
141        user_info.grant_privileges = get_user_privilege(user_info.id as _, &txn).await?;
142        txn.commit().await?;
143
144        let version = self
145            .notify_frontend(
146                NotificationOperation::Update,
147                NotificationInfo::User(user_info),
148            )
149            .await;
150
151        Ok(version)
152    }
153
154    #[cfg(test)]
155    pub async fn get_user(&self, id: UserId) -> MetaResult<user::Model> {
156        let inner = self.inner.read().await;
157        let user = User::find_by_id(id)
158            .one(&inner.db)
159            .await?
160            .ok_or_else(|| MetaError::catalog_id_not_found("user", id))?;
161        Ok(user)
162    }
163
164    #[cfg(test)]
165    pub async fn get_user_by_name(&self, name: &str) -> MetaResult<user::Model> {
166        let inner = self.inner.read().await;
167        let user = User::find()
168            .filter(user::Column::Name.eq(name))
169            .one(&inner.db)
170            .await?
171            .ok_or_else(|| anyhow::anyhow!("user {name} not found"))?;
172        Ok(user)
173    }
174
175    pub async fn drop_user(&self, user_id: UserId) -> MetaResult<NotificationVersion> {
176        let inner = self.inner.write().await;
177        let txn = inner.db.begin().await?;
178        let user = User::find_by_id(user_id)
179            .one(&txn)
180            .await?
181            .ok_or_else(|| MetaError::catalog_id_not_found("user", user_id))?;
182        if user.name == DEFAULT_SUPER_USER || user.name == DEFAULT_SUPER_USER_FOR_PG {
183            return Err(MetaError::permission_denied(format!(
184                "drop default super user {} is not allowed",
185                user.name
186            )));
187        }
188
189        // check if the user is the owner of any objects.
190        let count = Object::find()
191            .filter(object::Column::OwnerId.eq(user_id))
192            .count(&txn)
193            .await?;
194        if count != 0 {
195            return Err(MetaError::permission_denied(format!(
196                "drop user {} is not allowed, because it owns {} objects",
197                user.name, count
198            )));
199        }
200
201        // check if the user granted any privileges to other users.
202        let count = UserPrivilege::find()
203            .filter(user_privilege::Column::GrantedBy.eq(user_id))
204            .count(&txn)
205            .await?;
206        if count != 0 {
207            return Err(MetaError::permission_denied(format!(
208                "drop user {} is not allowed, because it granted {} privileges to others",
209                user.name, count
210            )));
211        }
212
213        let res = User::delete_by_id(user_id).exec(&txn).await?;
214        if res.rows_affected != 1 {
215            return Err(MetaError::catalog_id_not_found("user", user_id));
216        }
217        txn.commit().await?;
218
219        let version = self
220            .notify_frontend(
221                NotificationOperation::Delete,
222                NotificationInfo::User(PbUserInfo {
223                    id: user_id as _,
224                    ..Default::default()
225                }),
226            )
227            .await;
228
229        Ok(version)
230    }
231
232    pub async fn grant_privilege(
233        &self,
234        user_ids: Vec<UserId>,
235        new_grant_privileges: &[PbGrantPrivilege],
236        grantor: UserId,
237        with_grant_option: bool,
238    ) -> MetaResult<Option<NotificationVersion>> {
239        let inner = self.inner.write().await;
240        let txn = inner.db.begin().await?;
241        for user_id in &user_ids {
242            ensure_user_id(*user_id, &txn).await?;
243        }
244
245        let mut privileges = vec![];
246        for gp in new_grant_privileges {
247            let id = extract_grant_obj_id(gp.get_object()?);
248            let internal_table_ids = get_internal_tables_by_id(id.as_job_id(), &txn).await?;
249            let index_state_table_ids =
250                get_index_state_tables_by_table_id(id.as_table_id(), &txn).await?;
251            for action_with_opt in &gp.action_with_opts {
252                let action = action_with_opt.get_action()?.into();
253                privileges.push(user_privilege::ActiveModel {
254                    oid: Set(id),
255                    granted_by: Set(grantor),
256                    action: Set(action),
257                    with_grant_option: Set(with_grant_option),
258                    ..Default::default()
259                });
260                if action == Action::Select {
261                    privileges.extend(
262                        internal_table_ids
263                            .iter()
264                            .chain(index_state_table_ids.iter())
265                            .map(|&tid| user_privilege::ActiveModel {
266                                oid: Set(tid.as_object_id()),
267                                granted_by: Set(grantor),
268                                action: Set(Action::Select),
269                                with_grant_option: Set(with_grant_option),
270                                ..Default::default()
271                            }),
272                    );
273                    let iceberg_privilege_object_ids =
274                        get_iceberg_related_object_ids(id, &txn).await?;
275                    privileges.extend(iceberg_privilege_object_ids.iter().map(
276                        |&iceberg_object_id| user_privilege::ActiveModel {
277                            oid: Set(iceberg_object_id),
278                            granted_by: Set(grantor),
279                            action: Set(action),
280                            with_grant_option: Set(with_grant_option),
281                            ..Default::default()
282                        },
283                    ));
284                }
285            }
286        }
287
288        // check whether grantor has the privilege to grant the privilege.
289        let user = User::find_by_id(grantor)
290            .one(&txn)
291            .await?
292            .ok_or_else(|| MetaError::catalog_id_not_found("user", grantor))?;
293        let mut filtered_privileges = vec![];
294        if !user.is_super {
295            for mut privilege in privileges {
296                if grantor == get_object_owner(*privilege.oid.as_ref(), &txn).await? {
297                    filtered_privileges.push(privilege);
298                    continue;
299                }
300                let filter = user_privilege::Column::UserId
301                    .eq(grantor)
302                    .and(user_privilege::Column::Oid.eq(*privilege.oid.as_ref()))
303                    .and(user_privilege::Column::Action.eq(*privilege.action.as_ref()))
304                    .and(user_privilege::Column::WithGrantOption.eq(true));
305                let privilege_id: Option<PrivilegeId> = UserPrivilege::find()
306                    .select_only()
307                    .column(user_privilege::Column::Id)
308                    .filter(filter)
309                    .into_tuple()
310                    .one(&txn)
311                    .await?;
312                let Some(privilege_id) = privilege_id else {
313                    tracing::warn!(
314                        "user {} don't have privilege {:?} or grant option",
315                        grantor,
316                        privilege.action
317                    );
318                    continue;
319                };
320                privilege.dependent_id = Set(Some(privilege_id));
321                filtered_privileges.push(privilege);
322            }
323        } else {
324            filtered_privileges = privileges;
325        }
326
327        // insert privileges
328        let user_privileges = user_ids
329            .iter()
330            .flat_map(|user_id| {
331                filtered_privileges.iter().map(|p| {
332                    let mut p = p.clone();
333                    p.user_id = Set(*user_id);
334                    p
335                })
336            })
337            .collect_vec();
338        upsert_user_privileges(&txn, user_privileges).await?;
339
340        let user_infos = list_user_info_by_ids(user_ids, &txn).await?;
341
342        txn.commit().await?;
343
344        let version = self.notify_users_update(user_infos).await;
345        Ok(version)
346    }
347
348    pub async fn revoke_privilege(
349        &self,
350        user_ids: Vec<UserId>,
351        revoke_grant_privileges: &[PbGrantPrivilege],
352        granted_by: UserId,
353        revoke_by: UserId,
354        revoke_grant_option: bool,
355        cascade: bool,
356    ) -> MetaResult<Option<NotificationVersion>> {
357        let inner = self.inner.write().await;
358        let txn = inner.db.begin().await?;
359        for user_id in &user_ids {
360            ensure_user_id(*user_id, &txn).await?;
361        }
362        // check whether revoke has the privilege to grant the privilege.
363        let revoke_user = User::find_by_id(revoke_by)
364            .one(&txn)
365            .await?
366            .ok_or_else(|| MetaError::catalog_id_not_found("user", revoke_by))?;
367
368        // check whether user can revoke the privilege.
369        if !revoke_user.is_super && granted_by != revoke_by {
370            let granted_user_name: String = User::find_by_id(granted_by)
371                .select_only()
372                .column(user::Column::Name)
373                .into_tuple()
374                .one(&txn)
375                .await?
376                .ok_or_else(|| MetaError::catalog_id_not_found("user", granted_by))?;
377            return Err(MetaError::permission_denied(format!(
378                "user {} is not super, can't revoke privileges for {}",
379                revoke_user.name, granted_user_name
380            )));
381        }
382
383        let mut revoke_items = HashMap::new();
384        for privilege in revoke_grant_privileges {
385            let obj = extract_grant_obj_id(privilege.get_object()?);
386            let internal_table_ids = get_internal_tables_by_id(obj.as_job_id(), &txn).await?;
387            let index_state_table_ids =
388                get_index_state_tables_by_table_id(obj.as_table_id(), &txn).await?;
389            let mut include_select = false;
390            let actions = privilege
391                .action_with_opts
392                .iter()
393                .map(|ao| {
394                    let action = Action::from(ao.get_action().unwrap());
395                    if action == Action::Select {
396                        include_select = true;
397                    }
398                    action
399                })
400                .collect_vec();
401            revoke_items.insert(obj, actions);
402            if include_select {
403                revoke_items.extend(
404                    internal_table_ids
405                        .iter()
406                        .chain(index_state_table_ids.iter())
407                        .map(|&tid| (tid.as_object_id(), vec![Action::Select])),
408                );
409                let iceberg_privilege_object_ids =
410                    get_iceberg_related_object_ids(obj, &txn).await?;
411                if !iceberg_privilege_object_ids.is_empty() {
412                    revoke_items.extend(
413                        iceberg_privilege_object_ids
414                            .into_iter()
415                            .map(|iceberg_object_id| (iceberg_object_id, vec![Action::Select])),
416                    );
417                }
418            }
419        }
420
421        let filter = if !revoke_user.is_super {
422            // ensure user have grant options or is owner of the object.
423            for (obj, actions) in &revoke_items {
424                if revoke_by == get_object_owner(*obj, &txn).await? {
425                    continue;
426                }
427                let owned_actions: HashSet<Action> = UserPrivilege::find()
428                    .select_only()
429                    .column(user_privilege::Column::Action)
430                    .filter(
431                        user_privilege::Column::UserId
432                            .eq(granted_by)
433                            .and(user_privilege::Column::Oid.eq(*obj))
434                            .and(user_privilege::Column::WithGrantOption.eq(true)),
435                    )
436                    .into_tuple::<Action>()
437                    .all(&txn)
438                    .await?
439                    .into_iter()
440                    .collect();
441                if actions.iter().any(|ac| !owned_actions.contains(ac)) {
442                    return Err(MetaError::permission_denied(format!(
443                        "user {} don't have privileges {:?} or grant option",
444                        revoke_user.name, actions,
445                    )));
446                }
447            }
448
449            user_privilege::Column::GrantedBy
450                .eq(granted_by)
451                .and(user_privilege::Column::UserId.is_in(user_ids.clone()))
452        } else {
453            user_privilege::Column::UserId.is_in(user_ids.clone())
454        };
455        let mut root_user_privileges: Vec<PartialUserPrivilege> = vec![];
456        for (obj, actions) in revoke_items {
457            let filter = filter
458                .clone()
459                .and(user_privilege::Column::Oid.eq(obj))
460                .and(user_privilege::Column::Action.is_in(actions));
461            root_user_privileges.extend(
462                UserPrivilege::find()
463                    .select_only()
464                    .columns([user_privilege::Column::Id, user_privilege::Column::UserId])
465                    .filter(filter)
466                    .into_partial_model()
467                    .all(&txn)
468                    .await?,
469            );
470        }
471        if root_user_privileges.is_empty() {
472            tracing::warn!("no privilege to revoke, ignore it");
473            return Ok(None);
474        }
475
476        // check if the user granted any privileges to other users.
477        let root_privilege_ids = root_user_privileges.iter().map(|ur| ur.id).collect_vec();
478        let (all_privilege_ids, to_update_user_ids) = if !cascade {
479            ensure_privileges_not_referred(root_privilege_ids.clone(), &txn).await?;
480            (
481                root_privilege_ids.clone(),
482                root_user_privileges
483                    .iter()
484                    .map(|ur| ur.user_id)
485                    .collect_vec(),
486            )
487        } else {
488            let all_user_privileges =
489                get_referring_privileges_cascade(root_privilege_ids.clone(), &txn).await?;
490            (
491                all_user_privileges.iter().map(|ur| ur.id).collect_vec(),
492                all_user_privileges
493                    .iter()
494                    .map(|ur| ur.user_id)
495                    .collect_vec(),
496            )
497        };
498
499        if revoke_grant_option {
500            UserPrivilege::update_many()
501                .col_expr(
502                    user_privilege::Column::WithGrantOption,
503                    SimpleExpr::Value(Value::Bool(Some(false))),
504                )
505                .filter(
506                    user_privilege::Column::Id
507                        .is_in(all_privilege_ids)
508                        .and(user_privilege::Column::WithGrantOption.eq(true)),
509                )
510                .exec(&txn)
511                .await?;
512        } else {
513            // The dependent privileges will be deleted cascade.
514            UserPrivilege::delete_many()
515                .filter(user_privilege::Column::Id.is_in(root_privilege_ids))
516                .exec(&txn)
517                .await?;
518        }
519
520        let user_infos = list_user_info_by_ids(to_update_user_ids, &txn).await?;
521
522        txn.commit().await?;
523
524        Ok(self.notify_users_update(user_infos).await)
525    }
526
527    pub async fn grant_default_privileges(
528        &self,
529        user_ids: Vec<UserId>,
530        database_id: DatabaseId,
531        schema_ids: Vec<SchemaId>,
532        grantor: UserId,
533        actions: Vec<PbAction>,
534        object_type: PbObjectType,
535        grantees: Vec<UserId>,
536        with_grant_option: bool,
537    ) -> MetaResult<()> {
538        tracing::debug!(
539            ?user_ids,
540            %database_id,
541            ?schema_ids,
542            ?actions,
543            ?object_type,
544            ?grantees,
545            with_grant_option,
546            "grant default privileges",
547        );
548        let inner = self.inner.write().await;
549        let txn = inner.db.begin().await?;
550        for user_id in &user_ids {
551            ensure_user_id(*user_id, &txn).await?;
552        }
553        ensure_object_id(ObjectType::Database, database_id, &txn).await?;
554        for schema_id in &schema_ids {
555            ensure_object_id(ObjectType::Schema, *schema_id, &txn).await?;
556        }
557        for grantee in &grantees {
558            ensure_user_id(*grantee, &txn).await?;
559        }
560        if object_type == PbObjectType::Schema {
561            assert!(
562                schema_ids.is_empty(),
563                "schema_ids should be empty when object_type is Schema"
564            );
565
566            // Note that the UNIQUE constraint does not treat NULL values as equal, we cannot rely on conflict check
567            // to update it and have to check existing default privileges manually.
568            let actions = actions.iter().map(|&a| Action::from(a)).collect_vec();
569            let existing_default_privileges: HashMap<_, _> = UserDefaultPrivilege::find()
570                .select_only()
571                .columns([
572                    user_default_privilege::Column::Id,
573                    user_default_privilege::Column::UserId,
574                    user_default_privilege::Column::Grantee,
575                    user_default_privilege::Column::Action,
576                ])
577                .filter(
578                    user_default_privilege::Column::DatabaseId
579                        .eq(database_id)
580                        .and(user_default_privilege::Column::ObjectType.eq(ObjectType::Schema))
581                        .and(user_default_privilege::Column::UserId.is_in(user_ids.clone()))
582                        .and(user_default_privilege::Column::Grantee.is_in(grantees.clone()))
583                        .and(user_default_privilege::Column::Action.is_in(actions.clone())),
584                )
585                .into_tuple::<(DefaultPrivilegeId, UserId, UserId, Action)>()
586                .all(&txn)
587                .await?
588                .into_iter()
589                .map(|(id, user_id, grantee, action)| ((user_id, grantee, action), id))
590                .collect();
591
592            for user_id in user_ids {
593                for grantee in &grantees {
594                    for action in &actions {
595                        if let Some(existing_id) =
596                            existing_default_privileges.get(&(user_id, *grantee, *action))
597                            && with_grant_option
598                        {
599                            // If the default privilege already exists, we should update the grant option.
600                            UserDefaultPrivilege::update(user_default_privilege::ActiveModel {
601                                id: Set(*existing_id),
602                                with_grant_option: Set(true),
603                                granted_by: Set(grantor as _),
604                                ..Default::default()
605                            })
606                            .exec(&txn)
607                            .await?;
608                        } else {
609                            UserDefaultPrivilege::insert(user_default_privilege::ActiveModel {
610                                id: Default::default(),
611                                database_id: Set(database_id),
612                                schema_id: Set(None),
613                                object_type: Set(ObjectType::Schema),
614                                for_materialized_view: Set(false),
615                                user_id: Set(user_id),
616                                grantee: Set(*grantee),
617                                granted_by: Set(grantor as _),
618                                action: Set(*action),
619                                with_grant_option: Set(with_grant_option),
620                            })
621                            .exec(&txn)
622                            .await?;
623                        }
624                    }
625                }
626            }
627        } else {
628            let mut default_privileges = vec![];
629            for user_id in user_ids {
630                for grantee in &grantees {
631                    for action in &actions {
632                        if schema_ids.is_empty() {
633                            default_privileges.push(user_default_privilege::ActiveModel {
634                                id: Default::default(),
635                                database_id: Set(database_id),
636                                schema_id: Set(None),
637                                object_type: Set(object_type.into()),
638                                for_materialized_view: Set(object_type == PbObjectType::Mview),
639                                user_id: Set(user_id),
640                                grantee: Set(*grantee),
641                                granted_by: Set(grantor as _),
642                                action: Set((*action).into()),
643                                with_grant_option: Set(with_grant_option),
644                            });
645                            continue;
646                        }
647                        for schema_id in &schema_ids {
648                            default_privileges.push(user_default_privilege::ActiveModel {
649                                id: Default::default(),
650                                database_id: Set(database_id),
651                                schema_id: Set(Some(*schema_id)),
652                                object_type: Set(object_type.into()),
653                                for_materialized_view: Set(object_type == PbObjectType::Mview),
654                                user_id: Set(user_id),
655                                grantee: Set(*grantee),
656                                granted_by: Set(grantor as _),
657                                action: Set((*action).into()),
658                                with_grant_option: Set(with_grant_option),
659                            });
660                        }
661                    }
662                }
663            }
664
665            let mut on_conflict = OnConflict::columns([
666                user_default_privilege::Column::UserId,
667                user_default_privilege::Column::DatabaseId,
668                user_default_privilege::Column::SchemaId,
669                user_default_privilege::Column::ObjectType,
670                user_default_privilege::Column::ForMaterializedView,
671                user_default_privilege::Column::Grantee,
672                user_default_privilege::Column::Action,
673            ]);
674            if with_grant_option {
675                on_conflict.update_column(user_default_privilege::Column::WithGrantOption);
676            } else {
677                // Workaround to support MYSQL for `DO NOTHING`.
678                on_conflict.update_column(user_default_privilege::Column::UserId);
679            }
680            UserDefaultPrivilege::insert_many(default_privileges)
681                .on_conflict(on_conflict)
682                .do_nothing()
683                .exec(&txn)
684                .await?;
685        }
686
687        txn.commit().await?;
688        Ok(())
689    }
690
691    pub async fn revoke_default_privileges(
692        &self,
693        user_ids: Vec<UserId>,
694        database_id: DatabaseId,
695        schema_ids: Vec<SchemaId>,
696        actions: Vec<PbAction>,
697        object_type: PbObjectType,
698        grantees: Vec<UserId>,
699        revoke_grant_option: bool,
700    ) -> MetaResult<()> {
701        let inner = self.inner.write().await;
702        let txn = inner.db.begin().await?;
703        for user_id in &user_ids {
704            ensure_user_id(*user_id, &txn).await?;
705        }
706
707        let schema_filter = if schema_ids.is_empty() {
708            user_default_privilege::Column::SchemaId.is_null()
709        } else {
710            user_default_privilege::Column::SchemaId.is_in(schema_ids)
711        };
712        let filter = user_default_privilege::Column::DatabaseId
713            .eq(database_id)
714            .and(schema_filter)
715            .and(user_default_privilege::Column::UserId.is_in(user_ids))
716            .and(user_default_privilege::Column::ObjectType.eq(ObjectType::from(object_type)))
717            .and(user_default_privilege::Column::Grantee.is_in(grantees))
718            .and(
719                user_default_privilege::Column::Action
720                    .is_in(actions.iter().map(|&a| Action::from(a))),
721            );
722
723        if revoke_grant_option {
724            // update the `with_grant_option` field to false
725            let res = UserDefaultPrivilege::update_many()
726                .col_expr(
727                    user_default_privilege::Column::WithGrantOption,
728                    SimpleExpr::Value(Value::Bool(Some(false))),
729                )
730                .filter(filter.and(user_default_privilege::Column::WithGrantOption.eq(true)))
731                .exec(&txn)
732                .await?;
733            tracing::info!(
734                "revoke {count} grant option for default privileges",
735                count = res.rows_affected
736            );
737        } else {
738            let res = UserDefaultPrivilege::delete_many()
739                .filter(filter)
740                .exec(&txn)
741                .await?;
742            tracing::info!(
743                "revoke {count} default privileges",
744                count = res.rows_affected
745            );
746        }
747
748        txn.commit().await?;
749        Ok(())
750    }
751}
752
753#[cfg(test)]
754mod tests {
755    use risingwave_pb::user::grant_privilege::{PbActionWithGrantOption, PbObject};
756
757    use super::*;
758    use crate::manager::MetaSrvEnv;
759
760    const TEST_DATABASE_ID: DatabaseId = DatabaseId::new(1);
761    const TEST_ROOT_USER_ID: UserId = UserId::new(1);
762
763    fn make_test_user(name: &str) -> PbUserInfo {
764        PbUserInfo {
765            name: name.to_owned(),
766            ..Default::default()
767        }
768    }
769
770    fn make_privilege(
771        object: PbObject,
772        actions: &[PbAction],
773        with_grant_option: bool,
774    ) -> PbGrantPrivilege {
775        PbGrantPrivilege {
776            object: Some(object),
777            action_with_opts: actions
778                .iter()
779                .map(|&action| PbActionWithGrantOption {
780                    action: action as _,
781                    with_grant_option,
782                    ..Default::default()
783                })
784                .collect(),
785        }
786    }
787
788    #[tokio::test]
789    async fn test_user_and_privilege() -> MetaResult<()> {
790        let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
791        assert_eq!(
792            mgr.grant_privilege(vec![], &[], TEST_ROOT_USER_ID, false)
793                .await?,
794            None
795        );
796        mgr.create_user(make_test_user("test_user_1")).await?;
797        mgr.create_user(make_test_user("test_user_2")).await?;
798        let user_1 = mgr.get_user_by_name("test_user_1").await?;
799        let user_2 = mgr.get_user_by_name("test_user_2").await?;
800
801        assert!(
802            mgr.create_user(make_test_user("test_user_1"))
803                .await
804                .is_err(),
805            "user_1 already exists"
806        );
807        mgr.update_user(
808            PbUserInfo {
809                id: user_1.user_id as _,
810                name: "test_user_1_new".to_owned(),
811                ..Default::default()
812            },
813            &[PbUpdateField::Rename],
814        )
815        .await?;
816        let user_1 = mgr.get_user(user_1.user_id).await?;
817        assert_eq!(user_1.name, "test_user_1_new".to_owned());
818
819        let conn_with_option = make_privilege(TEST_DATABASE_ID.into(), &[PbAction::Connect], true);
820        let create_without_option =
821            make_privilege(TEST_DATABASE_ID.into(), &[PbAction::Create], false);
822        // ROOT grant CONN with grant option to user_1.
823        mgr.grant_privilege(
824            vec![user_1.user_id],
825            std::slice::from_ref(&conn_with_option),
826            TEST_ROOT_USER_ID,
827            true,
828        )
829        .await?;
830        // ROOT grant CREATE without grant option to user_1.
831        mgr.grant_privilege(
832            vec![user_1.user_id],
833            std::slice::from_ref(&create_without_option),
834            TEST_ROOT_USER_ID,
835            false,
836        )
837        .await?;
838        // user_1 grant CONN with grant option to user_2.
839        mgr.grant_privilege(
840            vec![user_2.user_id],
841            std::slice::from_ref(&conn_with_option),
842            user_1.user_id,
843            true,
844        )
845        .await?;
846        mgr.grant_privilege(
847            vec![user_2.user_id],
848            std::slice::from_ref(&create_without_option),
849            user_1.user_id,
850            false,
851        )
852        .await?;
853
854        assert!(
855            mgr.drop_user(user_1.user_id).await.is_err(),
856            "user_1 can't be dropped"
857        );
858
859        let privilege_1 = get_user_privilege(user_1.user_id, &mgr.inner.read().await.db).await?;
860        assert_eq!(privilege_1.len(), 2);
861        assert!(
862            privilege_1
863                .iter()
864                .all(|gp| gp.object == Some(TEST_DATABASE_ID.into())
865                    && gp.action_with_opts[0].granted_by == TEST_ROOT_USER_ID)
866        );
867
868        let privilege_2 = get_user_privilege(user_2.user_id, &mgr.inner.read().await.db).await?;
869        assert_eq!(privilege_2.len(), 1);
870        assert!(
871            privilege_2
872                .iter()
873                .all(|gp| gp.object == Some(TEST_DATABASE_ID.into())
874                    && gp.action_with_opts[0].granted_by == user_1.user_id
875                    && gp.action_with_opts[0].with_grant_option)
876        );
877
878        // revoke privilege for others by non-super user.
879        assert!(
880            mgr.revoke_privilege(
881                vec![user_1.user_id],
882                std::slice::from_ref(&conn_with_option),
883                TEST_ROOT_USER_ID,
884                user_2.user_id,
885                false,
886                false
887            )
888            .await
889            .is_err(),
890            "user_2 can't revoke for user_1"
891        );
892
893        // revoke privilege without grant option.
894        assert!(
895            mgr.revoke_privilege(
896                vec![user_2.user_id],
897                std::slice::from_ref(&create_without_option),
898                user_1.user_id,
899                user_1.user_id,
900                false,
901                false
902            )
903            .await
904            .is_err(),
905            "user_2 don't have grant option for CREATE"
906        );
907
908        // revoke referred privilege in restrict mode.
909        assert!(
910            mgr.revoke_privilege(
911                vec![user_1.user_id],
912                std::slice::from_ref(&conn_with_option),
913                TEST_ROOT_USER_ID,
914                TEST_ROOT_USER_ID,
915                false,
916                false
917            )
918            .await
919            .is_err(),
920            "permission deny in restrict mode, CONN granted to user_2"
921        );
922
923        // revoke non-referred privilege in restrict mode.
924        mgr.revoke_privilege(
925            vec![user_1.user_id],
926            std::slice::from_ref(&create_without_option),
927            TEST_ROOT_USER_ID,
928            TEST_ROOT_USER_ID,
929            false,
930            false,
931        )
932        .await?;
933
934        let privilege_1 = get_user_privilege(user_1.user_id, &mgr.inner.read().await.db).await?;
935        assert_eq!(privilege_1.len(), 1);
936        assert!(
937            privilege_1
938                .iter()
939                .all(|gp| gp.object == Some(TEST_DATABASE_ID.into())
940                    && gp.action_with_opts[0].action == PbAction::Connect as i32)
941        );
942
943        // revoke grant option for referred privilege in cascade mode.
944        mgr.revoke_privilege(
945            vec![user_1.user_id],
946            std::slice::from_ref(&conn_with_option),
947            TEST_ROOT_USER_ID,
948            TEST_ROOT_USER_ID,
949            true,
950            true,
951        )
952        .await?;
953        let privilege_1 = get_user_privilege(user_1.user_id, &mgr.inner.read().await.db).await?;
954        assert_eq!(privilege_1.len(), 1);
955        assert!(
956            privilege_1
957                .iter()
958                .all(|gp| gp.object == Some(TEST_DATABASE_ID.into())
959                    && gp.action_with_opts[0].action == PbAction::Connect as i32
960                    && !gp.action_with_opts[0].with_grant_option)
961        );
962        let privilege_2 = get_user_privilege(user_2.user_id, &mgr.inner.read().await.db).await?;
963        assert_eq!(privilege_2.len(), 1);
964        assert!(
965            privilege_2
966                .iter()
967                .all(|gp| gp.object == Some(TEST_DATABASE_ID.into())
968                    && gp.action_with_opts[0].action == PbAction::Connect as i32
969                    && !gp.action_with_opts[0].with_grant_option)
970        );
971
972        // revoke referred privilege in cascade mode.
973        mgr.revoke_privilege(
974            vec![user_1.user_id],
975            std::slice::from_ref(&conn_with_option),
976            TEST_ROOT_USER_ID,
977            TEST_ROOT_USER_ID,
978            false,
979            true,
980        )
981        .await?;
982        let privilege_1 = get_user_privilege(user_1.user_id, &mgr.inner.read().await.db).await?;
983        assert!(privilege_1.is_empty());
984        let privilege_2 = get_user_privilege(user_2.user_id, &mgr.inner.read().await.db).await?;
985        assert!(privilege_2.is_empty());
986
987        mgr.drop_user(user_1.user_id).await?;
988        mgr.drop_user(user_2.user_id).await?;
989        Ok(())
990    }
991}
risingwave_meta/controller/user.rs

risingwave_meta/controller/
user.rs
