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

risingwave_meta/controller/
user.rs
