uqa_sql/catalog/security/
table_grants.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Table and column GRANT/REVOKE validation, ACL candidates and diagnostics.
8#[cfg(test)]
9mod tests;
10use super::{
11    columns::{grant_column_acl, revoke_column_acl, select_column_acl_grantor},
12    table::{
13        grant_acl, revoke_acl, select_acl_grantor, validate_table_security_invariants,
14        RequestedTablePrivileges, TableAclPrivilege,
15    },
16    BoundTableSecurity, TableSecurity,
17};
18use crate::catalog::roles::identity::RoleSubject;
19use crate::catalog::{
20    roles::{RoleDefinition, RoleMembership, RoleMembershipKey},
21    stored_view::StoredView,
22};
23use crate::{
24    ast::{GrantTableStmt, SequencePrivilege, TablePrivilege, TableRevokeBehavior},
25    SQLError,
26};
27use std::collections::{BTreeMap, BTreeSet};
28use uqa_core::catalog_acl::AclGrantee;
29use uqa_core::RelationIdentity;
30pub type ViewPrivilegeUpdate = (RelationIdentity, StoredView);
31pub type ForeignTablePrivilegeUpdate = (RelationIdentity, BoundTableSecurity);
32pub type ForeignTableGrantTarget<'a> = (
33    &'a ResolvedTableGrantTarget,
34    BoundTableSecurity,
35    Vec<String>,
36);
37pub mod targets;
38pub struct ResolvedTableGrantTarget {
39    pub requested: String,
40    pub name: String,
41    pub relation: RelationIdentity,
42    pub kind: &'static str,
43    /// Attribute tuples selected in column order before authorization or writer waits. `None` denotes an uncoordinated analysis input.
44    pub acl_columns: Option<BTreeSet<String>>,
45}
46impl ResolvedTableGrantTarget {
47    pub fn includes_acl_tuple(&self, column: Option<&str>) -> bool {
48        column.is_none_or(|column| {
49            self.acl_columns
50                .as_ref()
51                .is_none_or(|columns| columns.contains(column))
52        })
53    }
54}
55
56fn apply_table_acl(
57    statement: &GrantTableStmt,
58    grantees: &[AclGrantee],
59    privileges: &[TableAclPrivilege],
60    current_user: &(impl RoleSubject + ?Sized),
61    roles: &BTreeMap<String, RoleDefinition>,
62    memberships: &BTreeMap<RoleMembershipKey, RoleMembership>,
63    current: &TableSecurity,
64) -> Result<(TableSecurity, usize), SQLError> {
65    let grantors = privileges
66        .iter()
67        .map(|privilege| {
68            (
69                *privilege,
70                select_acl_grantor(current, *privilege, current_user, roles, memberships),
71            )
72        })
73        .collect::<Vec<_>>();
74    let grantable = grantors
75        .iter()
76        .filter(|(_, grantor)| grantor.is_some())
77        .count();
78    let mut next = current.clone();
79    for (privilege, grantor) in grantors {
80        let Some(grantor) = grantor else {
81            continue;
82        };
83        if statement.is_grant {
84            grant_acl(
85                &mut next,
86                privilege,
87                grantees,
88                &grantor,
89                statement.grant_option,
90            );
91        } else {
92            revoke_acl(
93                &mut next,
94                privilege,
95                grantees,
96                &grantor,
97                statement.grant_option_only,
98                statement.revoke_behavior == TableRevokeBehavior::Cascade,
99            )?;
100        }
101    }
102    Ok((next, grantable))
103}
104
105fn apply_column_acl(
106    application: &TableGrantApplication<'_>,
107    privileges: &[(TableAclPrivilege, String)],
108    authorization: &TableSecurity,
109    current: &TableSecurity,
110    selected: Option<&BTreeSet<String>>,
111) -> Result<(TableSecurity, usize), SQLError> {
112    let TableGrantApplication {
113        statement,
114        grantees,
115        current_user,
116        roles,
117        memberships,
118        ..
119    } = application;
120    let grantors = privileges
121        .iter()
122        .map(|(privilege, column)| {
123            (
124                *privilege,
125                column.clone(),
126                select_column_acl_grantor(
127                    authorization,
128                    column,
129                    *privilege,
130                    current_user,
131                    roles,
132                    memberships,
133                ),
134            )
135        })
136        .collect::<Vec<_>>();
137    let grantable = grantors
138        .iter()
139        .filter(|(privilege, column, grantor)| {
140            grantor.is_some()
141                && application
142                    .requested
143                    .columns
144                    .contains(&(*privilege, column.clone()))
145        })
146        .count();
147    let mut next = current.clone();
148    for (privilege, column, grantor) in grantors {
149        if selected.is_some_and(|columns| !columns.contains(&column)) {
150            continue;
151        }
152        let Some(grantor) = grantor else {
153            continue;
154        };
155        if statement.is_grant {
156            grant_column_acl(
157                &mut next,
158                &column,
159                privilege,
160                grantees,
161                &grantor,
162                statement.grant_option,
163            );
164        } else {
165            revoke_column_acl(
166                &mut next,
167                &column,
168                privilege,
169                grantees,
170                &grantor,
171                statement.grant_option_only,
172                statement.revoke_behavior == TableRevokeBehavior::Cascade,
173            )?;
174        }
175    }
176    next.column_acls.retain(|_, acl| !acl.is_empty());
177    Ok((next, grantable))
178}
179
180pub struct TableGrantApplication<'a> {
181    pub statement: &'a GrantTableStmt,
182    pub grantees: &'a [AclGrantee],
183    pub requested: &'a RequestedTablePrivileges,
184    pub current_user: &'a dyn RoleSubject,
185    pub roles: &'a BTreeMap<String, RoleDefinition>,
186    pub memberships: &'a BTreeMap<RoleMembershipKey, RoleMembership>,
187}
188
189impl TableGrantApplication<'_> {
190    /// `PostgreSQL` replaces relation ACL tuples for table-level commands, and nonempty requested attribute ACLs even when their bits are unchanged.
191    pub fn replaced_tuples(
192        &self,
193        before: &TableSecurity,
194        after: &TableSecurity,
195    ) -> Vec<Option<String>> {
196        let mut tuples = Vec::new();
197        let implicit_columns = !self.statement.is_grant
198            && self.requested.table.iter().any(|privilege| {
199                matches!(
200                    privilege,
201                    TableAclPrivilege::Select
202                        | TableAclPrivilege::Insert
203                        | TableAclPrivilege::Update
204                        | TableAclPrivilege::References
205                )
206            });
207        if !self.requested.table.is_empty() {
208            tuples.push(None);
209        }
210        let columns = before
211            .column_acls
212            .keys()
213            .chain(after.column_acls.keys())
214            .chain(self.requested.columns.iter().map(|(_, column)| column))
215            .collect::<std::collections::BTreeSet<_>>();
216        for column in columns {
217            if before.column_acls.get(column) != after.column_acls.get(column)
218                || ((implicit_columns
219                    || self
220                        .requested
221                        .columns
222                        .iter()
223                        .any(|(_, name)| name == column))
224                    && after
225                        .column_acls
226                        .get(column)
227                        .is_some_and(|acl| !acl.is_empty()))
228            {
229                tuples.push(Some(column.clone()));
230            }
231        }
232        tuples
233    }
234
235    pub fn apply(&self, current: &TableSecurity) -> Result<(TableSecurity, usize), SQLError> {
236        self.apply_columns(current, None)
237    }
238
239    pub fn apply_to(
240        &self,
241        target: &ResolvedTableGrantTarget,
242        current: &TableSecurity,
243    ) -> Result<(TableSecurity, usize), SQLError> {
244        self.apply_columns(current, target.acl_columns.as_ref())
245    }
246
247    /// Inspect one attribute in catalog order without revisiting earlier attribute ACLs.
248    pub fn replaces_attribute(
249        &self,
250        current: &TableSecurity,
251        column: &str,
252    ) -> Result<bool, SQLError> {
253        let selected = BTreeSet::from([column.to_owned()]);
254        let (next, _) = self.apply_columns(current, Some(&selected))?;
255        Ok(self
256            .replaced_tuples(current, &next)
257            .iter()
258            .any(|tuple| tuple.as_deref() == Some(column)))
259    }
260
261    fn apply_columns(
262        &self,
263        current: &TableSecurity,
264        selected: Option<&BTreeSet<String>>,
265    ) -> Result<(TableSecurity, usize), SQLError> {
266        let (next, table_grantable) = apply_table_acl(
267            self.statement,
268            self.grantees,
269            &self.requested.table,
270            self.current_user,
271            self.roles,
272            self.memberships,
273            current,
274        )?;
275        let mut columns = self.requested.columns.clone();
276        let mut implied = Vec::new();
277        if !self.statement.is_grant {
278            for privilege in &self.requested.table {
279                if matches!(
280                    privilege,
281                    TableAclPrivilege::Select
282                        | TableAclPrivilege::Insert
283                        | TableAclPrivilege::Update
284                        | TableAclPrivilege::References
285                ) {
286                    for column in current.column_acls.keys() {
287                        let key = (*privilege, column.clone());
288                        if !columns.contains(&key) {
289                            columns.push(key.clone());
290                        }
291                        implied.push(key);
292                    }
293                }
294            }
295        }
296        let (mut next, column_grantable) =
297            apply_column_acl(self, &columns, current, &next, selected)?;
298        // Relation grant-option loss also invalidates column grants made through that relation authority.
299        for (privilege, column) in implied {
300            if selected.is_some_and(|columns| !columns.contains(&column)) {
301                continue;
302            }
303            let before = super::columns::column_grant_option_roles(current, &column, privilege);
304            super::columns::revoke_dependent_column_acl(
305                &mut next,
306                &column,
307                privilege,
308                &before,
309                self.statement.revoke_behavior == TableRevokeBehavior::Cascade,
310            )?;
311        }
312        next.column_acls.retain(|_, acl| !acl.is_empty());
313        Ok((next, table_grantable + column_grantable))
314    }
315
316    pub fn record_warning(
317        &self,
318        grantable: usize,
319        relation: &RelationIdentity,
320        notices: &mut Vec<(&'static str, String)>,
321    ) {
322        let requested = self.requested.table.len() + self.requested.columns.len();
323        if grantable != requested {
324            notices.push(table_acl_warning(
325                self.statement.is_grant,
326                grantable != 0,
327                &relation.name,
328            ));
329        }
330    }
331}
332
333pub fn validate_requested_columns(
334    target: &RelationIdentity,
335    columns: &[String],
336    requested: &RequestedTablePrivileges,
337) -> Result<(), SQLError> {
338    for (_, requested_column) in &requested.columns {
339        if !columns.contains(requested_column) {
340            return Err(SQLError::Routine {
341                sqlstate: "42703".into(),
342                message: format!(
343                    "column \"{requested_column}\" of relation \"{}\" does not exist",
344                    target.name
345                ),
346            });
347        }
348    }
349    Ok(())
350}
351
352pub fn validate_table_grant_target_kinds(
353    statement: &GrantTableStmt,
354    targets: &[ResolvedTableGrantTarget],
355) -> Result<(), SQLError> {
356    for target in targets {
357        if !matches!(
358            target.kind,
359            "table" | "view" | "materialized view" | "foreign table" | "sequence"
360        ) {
361            return Err(SQLError::Unsupported(format!(
362                "{} privileges for \"{}\" are not supported",
363                target.kind, target.requested
364            )));
365        }
366    }
367    if let Some(column) = statement
368        .privileges
369        .iter()
370        .flat_map(|privilege| &privilege.columns)
371        .next()
372    {
373        if let Some(target) = targets.iter().find(|target| target.kind == "sequence") {
374            return Err(SQLError::Routine {
375                sqlstate: "42703".into(),
376                message: format!(
377                    "column \"{column}\" of relation \"{}\" does not exist",
378                    target.relation.name
379                ),
380            });
381        }
382    }
383    Ok(())
384}
385
386pub fn validate_table_acl_roles(
387    statement: &GrantTableStmt,
388    grantees: &[AclGrantee],
389    requested_grantor: Option<&str>,
390    current_user: &(impl RoleSubject + ?Sized),
391    roles: &BTreeMap<String, RoleDefinition>,
392) -> Result<(), SQLError> {
393    for role in grantees {
394        if role
395            .role_name()
396            .is_some_and(|name| !roles.contains_key(name))
397        {
398            return Err(SQLError::Routine {
399                sqlstate: "42704".into(),
400                message: format!("role \"{role}\" does not exist"),
401            });
402        }
403    }
404    if statement.is_grant && statement.grant_option && grantees.iter().any(AclGrantee::is_public) {
405        return Err(SQLError::Routine {
406            sqlstate: "0LP01".into(),
407            message: "grant options can only be granted to roles".into(),
408        });
409    }
410    if let Some(requested_grantor) = requested_grantor {
411        if !roles.contains_key(requested_grantor) {
412            return Err(SQLError::Routine {
413                sqlstate: "42704".into(),
414                message: format!("role \"{requested_grantor}\" does not exist"),
415            });
416        }
417        if current_user.role_name(roles) != Some(requested_grantor) {
418            return Err(SQLError::Routine {
419                sqlstate: "0A000".into(),
420                message: "grantor must be current user".into(),
421            });
422        }
423    }
424    Ok(())
425}
426
427pub fn table_sequence_privileges(
428    privileges: &[crate::ast::TablePrivilegeSpec],
429) -> (Vec<SequencePrivilege>, bool) {
430    if privileges.is_empty() {
431        return (
432            vec![
433                SequencePrivilege::Select,
434                SequencePrivilege::Update,
435                SequencePrivilege::Usage,
436            ],
437            false,
438        );
439    }
440    let mut mapped = Vec::new();
441    let mut inapplicable = false;
442    for spec in privileges {
443        let privilege = if spec.columns.is_empty() {
444            match &spec.privilege {
445                TablePrivilege::Select => Some(SequencePrivilege::Select),
446                TablePrivilege::Update => Some(SequencePrivilege::Update),
447                TablePrivilege::Usage => Some(SequencePrivilege::Usage),
448                _ => {
449                    inapplicable = true;
450                    None
451                }
452            }
453        } else {
454            Some(SequencePrivilege::ColumnsUnsupported)
455        };
456        if let Some(privilege) = privilege {
457            if !mapped.contains(&privilege) {
458                mapped.push(privilege);
459            }
460        }
461    }
462    (mapped, inapplicable)
463}
464
465fn table_acl_warning(is_grant: bool, partial: bool, name: &str) -> (&'static str, String) {
466    let message = match (is_grant, partial) {
467        (true, true) => format!("not all privileges were granted for \"{name}\""),
468        (true, false) => format!("no privileges were granted for \"{name}\""),
469        (false, true) => format!("not all privileges could be revoked for \"{name}\""),
470        (false, false) => format!("no privileges could be revoked for \"{name}\""),
471    };
472    ("WARNING", message)
473}
474pub fn view_privilege_updates(
475    targets: Vec<(&ResolvedTableGrantTarget, StoredView)>,
476    application: &TableGrantApplication<'_>,
477    notices: &mut Vec<(&'static str, String)>,
478    dependencies: &mut std::collections::BTreeSet<String>,
479) -> Result<Vec<ViewPrivilegeUpdate>, SQLError> {
480    let mut updates = Vec::new();
481    for (target, mut view) in targets {
482        let current = view
483            .security
484            .resolve(application.roles)
485            .map_err(SQLError::Internal)?;
486        let (next, grantable) = application.apply_to(target, &current)?;
487        crate::catalog::security::dependencies::added_table_acl_roles(
488            &current,
489            &next,
490            dependencies,
491        );
492        let columns = view.output_columns.as_deref().ok_or_else(|| {
493            SQLError::Internal(format!(
494                "loaded view `{}` has no durable public column metadata",
495                target.relation.qualified_name()
496            ))
497        })?;
498        validate_table_security_invariants(&next, Some(columns), application.roles).map_err(
499            |error| {
500                SQLError::Internal(format!(
501                    "view `{}` produced invalid privilege metadata: {error}",
502                    target.relation.qualified_name()
503                ))
504            },
505        )?;
506        application.record_warning(grantable, &target.relation, notices);
507        if application
508            .replaced_tuples(&current, &next)
509            .iter()
510            .any(|column| target.includes_acl_tuple(column.as_deref()))
511        {
512            view.set_security(
513                BoundTableSecurity::bind(&next, application.roles).map_err(SQLError::Internal)?,
514            );
515            updates.push((target.relation.clone(), view));
516        }
517    }
518    Ok(updates)
519}
520pub fn foreign_table_privilege_updates(
521    targets: Vec<ForeignTableGrantTarget<'_>>,
522    application: &TableGrantApplication<'_>,
523    notices: &mut Vec<(&'static str, String)>,
524    dependencies: &mut std::collections::BTreeSet<String>,
525) -> Result<Vec<ForeignTablePrivilegeUpdate>, SQLError> {
526    let mut updates = Vec::new();
527    for (target, current, columns) in targets {
528        let current = current
529            .resolve(application.roles)
530            .map_err(SQLError::Internal)?;
531        let (next, grantable) = application.apply_to(target, &current)?;
532        crate::catalog::security::dependencies::added_table_acl_roles(
533            &current,
534            &next,
535            dependencies,
536        );
537        validate_table_security_invariants(&next, Some(&columns), application.roles).map_err(
538            |error| {
539                SQLError::Internal(format!(
540                    "foreign table `{}` produced invalid privilege metadata: {error}",
541                    target.relation.qualified_name()
542                ))
543            },
544        )?;
545        application.record_warning(grantable, &target.relation, notices);
546        if application
547            .replaced_tuples(&current, &next)
548            .iter()
549            .any(|column| target.includes_acl_tuple(column.as_deref()))
550        {
551            updates.push((
552                target.relation.clone(),
553                BoundTableSecurity::bind(&next, application.roles).map_err(SQLError::Internal)?,
554            ));
555        }
556    }
557    Ok(updates)
558}
uqa_sql/catalog/security/table_grants.rs

uqa_sql/catalog/security/
table_grants.rs
