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<crate::SQLNotice>,
321    ) {
322        let requested = self.requested.table.len() + self.requested.columns.len();
323        if grantable != requested {
324            notices.push(super::acl_warning::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        // `ExecGrant_Relation` refuses indexes and composite types by their relation name.
358        let refused = match target.kind {
359            "index" => Some("an index"),
360            "composite type" => Some("a composite type"),
361            _ => None,
362        };
363        if let Some(refused) = refused {
364            return Err(SQLError::Routine {
365                sqlstate: "42809".into(),
366                message: format!("\"{}\" is {refused}", target.relation.name),
367            });
368        }
369        if !matches!(
370            target.kind,
371            "table" | "view" | "materialized view" | "foreign table" | "sequence"
372        ) {
373            return Err(SQLError::Unsupported(format!(
374                "{} privileges for \"{}\" are not supported",
375                target.kind, target.requested
376            )));
377        }
378    }
379    if let Some(column) = statement
380        .privileges
381        .iter()
382        .flat_map(|privilege| &privilege.columns)
383        .next()
384    {
385        if let Some(target) = targets.iter().find(|target| target.kind == "sequence") {
386            return Err(SQLError::Routine {
387                sqlstate: "42703".into(),
388                message: format!(
389                    "column \"{column}\" of relation \"{}\" does not exist",
390                    target.relation.name
391                ),
392            });
393        }
394    }
395    Ok(())
396}
397
398pub fn validate_table_acl_roles(
399    statement: &GrantTableStmt,
400    grantees: &[AclGrantee],
401    requested_grantor: Option<&str>,
402    current_user: &(impl RoleSubject + ?Sized),
403    roles: &BTreeMap<String, RoleDefinition>,
404) -> Result<(), SQLError> {
405    for role in grantees {
406        if role
407            .role_name()
408            .is_some_and(|name| !roles.contains_key(name))
409        {
410            return Err(SQLError::Routine {
411                sqlstate: "42704".into(),
412                message: format!("role \"{role}\" does not exist"),
413            });
414        }
415    }
416    if statement.is_grant && statement.grant_option && grantees.iter().any(AclGrantee::is_public) {
417        return Err(SQLError::Routine {
418            sqlstate: "0LP01".into(),
419            message: "grant options can only be granted to roles".into(),
420        });
421    }
422    if let Some(requested_grantor) = requested_grantor {
423        if !roles.contains_key(requested_grantor) {
424            return Err(SQLError::Routine {
425                sqlstate: "42704".into(),
426                message: format!("role \"{requested_grantor}\" does not exist"),
427            });
428        }
429        if current_user.role_name(roles) != Some(requested_grantor) {
430            return Err(SQLError::Routine {
431                sqlstate: "0A000".into(),
432                message: "grantor must be current user".into(),
433            });
434        }
435    }
436    Ok(())
437}
438
439pub fn table_sequence_privileges(
440    privileges: &[crate::ast::TablePrivilegeSpec],
441) -> (Vec<SequencePrivilege>, bool) {
442    if privileges.is_empty() {
443        return (
444            vec![
445                SequencePrivilege::Select,
446                SequencePrivilege::Update,
447                SequencePrivilege::Usage,
448            ],
449            false,
450        );
451    }
452    let mut mapped = Vec::new();
453    let mut inapplicable = false;
454    for spec in privileges {
455        let privilege = if spec.columns.is_empty() {
456            match &spec.privilege {
457                TablePrivilege::Select => Some(SequencePrivilege::Select),
458                TablePrivilege::Update => Some(SequencePrivilege::Update),
459                TablePrivilege::Usage => Some(SequencePrivilege::Usage),
460                _ => {
461                    inapplicable = true;
462                    None
463                }
464            }
465        } else {
466            Some(SequencePrivilege::ColumnsUnsupported)
467        };
468        if let Some(privilege) = privilege {
469            if !mapped.contains(&privilege) {
470                mapped.push(privilege);
471            }
472        }
473    }
474    (mapped, inapplicable)
475}
476
477pub fn view_privilege_updates(
478    targets: Vec<(&ResolvedTableGrantTarget, StoredView)>,
479    application: &TableGrantApplication<'_>,
480    notices: &mut Vec<crate::SQLNotice>,
481    dependencies: &mut std::collections::BTreeSet<String>,
482) -> Result<Vec<ViewPrivilegeUpdate>, SQLError> {
483    let mut updates = Vec::new();
484    for (target, mut view) in targets {
485        let current = view
486            .security
487            .resolve(application.roles)
488            .map_err(SQLError::Internal)?;
489        let (next, grantable) = application.apply_to(target, &current)?;
490        crate::catalog::security::dependencies::added_table_acl_roles(
491            &current,
492            &next,
493            dependencies,
494        );
495        let columns = view.output_columns.as_deref().ok_or_else(|| {
496            SQLError::Internal(format!(
497                "loaded view `{}` has no durable public column metadata",
498                target.relation.qualified_name()
499            ))
500        })?;
501        validate_table_security_invariants(&next, Some(columns), application.roles).map_err(
502            |error| {
503                SQLError::Internal(format!(
504                    "view `{}` produced invalid privilege metadata: {error}",
505                    target.relation.qualified_name()
506                ))
507            },
508        )?;
509        application.record_warning(grantable, &target.relation, notices);
510        if application
511            .replaced_tuples(&current, &next)
512            .iter()
513            .any(|column| target.includes_acl_tuple(column.as_deref()))
514        {
515            view.set_security(
516                BoundTableSecurity::bind(&next, application.roles).map_err(SQLError::Internal)?,
517            );
518            updates.push((target.relation.clone(), view));
519        }
520    }
521    Ok(updates)
522}
523pub fn foreign_table_privilege_updates(
524    targets: Vec<ForeignTableGrantTarget<'_>>,
525    application: &TableGrantApplication<'_>,
526    notices: &mut Vec<crate::SQLNotice>,
527    dependencies: &mut std::collections::BTreeSet<String>,
528) -> Result<Vec<ForeignTablePrivilegeUpdate>, SQLError> {
529    let mut updates = Vec::new();
530    for (target, current, columns) in targets {
531        let current = current
532            .resolve(application.roles)
533            .map_err(SQLError::Internal)?;
534        let (next, grantable) = application.apply_to(target, &current)?;
535        crate::catalog::security::dependencies::added_table_acl_roles(
536            &current,
537            &next,
538            dependencies,
539        );
540        validate_table_security_invariants(&next, Some(&columns), application.roles).map_err(
541            |error| {
542                SQLError::Internal(format!(
543                    "foreign table `{}` produced invalid privilege metadata: {error}",
544                    target.relation.qualified_name()
545                ))
546            },
547        )?;
548        application.record_warning(grantable, &target.relation, notices);
549        if application
550            .replaced_tuples(&current, &next)
551            .iter()
552            .any(|column| target.includes_acl_tuple(column.as_deref()))
553        {
554            updates.push((
555                target.relation.clone(),
556                BoundTableSecurity::bind(&next, application.roles).map_err(SQLError::Internal)?,
557            ));
558        }
559    }
560    Ok(updates)
561}
uqa_sql/catalog/security/table_grants.rs

uqa_sql/catalog/security/
table_grants.rs
