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    TableSecurity,
17};
18use crate::catalog::{
19    roles::{RoleDefinition, RoleMembership, RoleMembershipKey},
20    stored_view::StoredView,
21};
22use crate::{
23    ast::{GrantTableStmt, SequencePrivilege, TablePrivilege, TableRevokeBehavior},
24    SQLError,
25};
26use std::collections::BTreeMap;
27use uqa_core::RelationIdentity;
28pub type ViewPrivilegeUpdate = (RelationIdentity, StoredView);
29pub type ForeignTablePrivilegeUpdate = (RelationIdentity, TableSecurity);
30pub type ForeignTableGrantTarget<'a> = (&'a ResolvedTableGrantTarget, TableSecurity, Vec<String>);
31pub mod targets;
32pub struct ResolvedTableGrantTarget {
33    pub requested: String,
34    pub name: String,
35    pub relation: RelationIdentity,
36    pub kind: &'static str,
37}
38
39fn apply_table_acl(
40    statement: &GrantTableStmt,
41    grantees: &[String],
42    privileges: &[TableAclPrivilege],
43    current_user: &str,
44    roles: &BTreeMap<String, RoleDefinition>,
45    memberships: &BTreeMap<RoleMembershipKey, RoleMembership>,
46    current: &TableSecurity,
47) -> Result<(TableSecurity, usize), SQLError> {
48    let grantors = privileges
49        .iter()
50        .map(|privilege| {
51            (
52                *privilege,
53                select_acl_grantor(current, *privilege, current_user, roles, memberships),
54            )
55        })
56        .collect::<Vec<_>>();
57    let grantable = grantors
58        .iter()
59        .filter(|(_, grantor)| grantor.is_some())
60        .count();
61    let mut next = current.clone();
62    for (privilege, grantor) in grantors {
63        let Some(grantor) = grantor else {
64            continue;
65        };
66        if statement.is_grant {
67            grant_acl(
68                &mut next,
69                privilege,
70                grantees,
71                &grantor,
72                statement.grant_option,
73            );
74        } else {
75            revoke_acl(
76                &mut next,
77                privilege,
78                grantees,
79                &grantor,
80                statement.grant_option_only,
81                statement.revoke_behavior == TableRevokeBehavior::Cascade,
82            )?;
83        }
84    }
85    Ok((next, grantable))
86}
87
88fn apply_column_acl(
89    statement: &GrantTableStmt,
90    grantees: &[String],
91    privileges: &[(TableAclPrivilege, String)],
92    current_user: &str,
93    roles: &BTreeMap<String, RoleDefinition>,
94    memberships: &BTreeMap<RoleMembershipKey, RoleMembership>,
95    current: &TableSecurity,
96) -> Result<(TableSecurity, usize), SQLError> {
97    let grantors = privileges
98        .iter()
99        .map(|(privilege, column)| {
100            (
101                *privilege,
102                column.clone(),
103                select_column_acl_grantor(
104                    current,
105                    column,
106                    *privilege,
107                    current_user,
108                    roles,
109                    memberships,
110                ),
111            )
112        })
113        .collect::<Vec<_>>();
114    let grantable = grantors
115        .iter()
116        .filter(|(_, _, grantor)| grantor.is_some())
117        .count();
118    let mut next = current.clone();
119    for (privilege, column, grantor) in grantors {
120        let Some(grantor) = grantor else {
121            continue;
122        };
123        if statement.is_grant {
124            grant_column_acl(
125                &mut next,
126                &column,
127                privilege,
128                grantees,
129                &grantor,
130                statement.grant_option,
131            );
132        } else {
133            revoke_column_acl(
134                &mut next,
135                &column,
136                privilege,
137                grantees,
138                &grantor,
139                statement.grant_option_only,
140                statement.revoke_behavior == TableRevokeBehavior::Cascade,
141            )?;
142        }
143    }
144    next.column_acls.retain(|_, acl| !acl.is_empty());
145    Ok((next, grantable))
146}
147
148pub struct TableGrantApplication<'a> {
149    pub statement: &'a GrantTableStmt,
150    pub grantees: &'a [String],
151    pub requested: &'a RequestedTablePrivileges,
152    pub current_user: &'a str,
153    pub roles: &'a BTreeMap<String, RoleDefinition>,
154    pub memberships: &'a BTreeMap<RoleMembershipKey, RoleMembership>,
155}
156
157impl TableGrantApplication<'_> {
158    pub fn apply(&self, current: &TableSecurity) -> Result<(TableSecurity, usize), SQLError> {
159        let (next, table_grantable) = apply_table_acl(
160            self.statement,
161            self.grantees,
162            &self.requested.table,
163            self.current_user,
164            self.roles,
165            self.memberships,
166            current,
167        )?;
168        let (next, column_grantable) = apply_column_acl(
169            self.statement,
170            self.grantees,
171            &self.requested.columns,
172            self.current_user,
173            self.roles,
174            self.memberships,
175            &next,
176        )?;
177        Ok((next, table_grantable + column_grantable))
178    }
179
180    pub fn record_warning(
181        &self,
182        grantable: usize,
183        relation: &RelationIdentity,
184        notices: &mut Vec<(&'static str, String)>,
185    ) {
186        let requested = self.requested.table.len() + self.requested.columns.len();
187        if grantable != requested {
188            notices.push(table_acl_warning(
189                self.statement.is_grant,
190                grantable != 0,
191                &relation.name,
192            ));
193        }
194    }
195}
196
197pub fn validate_requested_columns(
198    target: &RelationIdentity,
199    columns: &[String],
200    requested: &RequestedTablePrivileges,
201) -> Result<(), SQLError> {
202    for (_, requested_column) in &requested.columns {
203        if !columns.contains(requested_column) {
204            return Err(SQLError::Routine {
205                sqlstate: "42703".into(),
206                message: format!(
207                    "column \"{requested_column}\" of relation \"{}\" does not exist",
208                    target.name
209                ),
210            });
211        }
212    }
213    Ok(())
214}
215
216pub fn validate_table_grant_target_kinds(
217    statement: &GrantTableStmt,
218    targets: &[ResolvedTableGrantTarget],
219) -> Result<(), SQLError> {
220    for target in targets {
221        if !matches!(
222            target.kind,
223            "table" | "view" | "materialized view" | "foreign table" | "sequence"
224        ) {
225            return Err(SQLError::Unsupported(format!(
226                "{} privileges for \"{}\" are not supported",
227                target.kind, target.requested
228            )));
229        }
230    }
231    if let Some(column) = statement
232        .privileges
233        .iter()
234        .flat_map(|privilege| &privilege.columns)
235        .next()
236    {
237        if let Some(target) = targets.iter().find(|target| target.kind == "sequence") {
238            return Err(SQLError::Routine {
239                sqlstate: "42703".into(),
240                message: format!(
241                    "column \"{column}\" of relation \"{}\" does not exist",
242                    target.relation.name
243                ),
244            });
245        }
246    }
247    Ok(())
248}
249
250pub fn validate_table_acl_roles(
251    statement: &GrantTableStmt,
252    grantees: &[String],
253    requested_grantor: Option<&str>,
254    current_user: &str,
255    roles: &BTreeMap<String, RoleDefinition>,
256) -> Result<(), SQLError> {
257    for role in grantees {
258        if role != "PUBLIC" && !roles.contains_key(role) {
259            return Err(SQLError::Routine {
260                sqlstate: "42704".into(),
261                message: format!("role \"{role}\" does not exist"),
262            });
263        }
264    }
265    if statement.is_grant && statement.grant_option && grantees.iter().any(|role| role == "PUBLIC")
266    {
267        return Err(SQLError::Routine {
268            sqlstate: "0LP01".into(),
269            message: "grant options can only be granted to roles".into(),
270        });
271    }
272    if let Some(requested_grantor) = requested_grantor {
273        if !roles.contains_key(requested_grantor) {
274            return Err(SQLError::Routine {
275                sqlstate: "42704".into(),
276                message: format!("role \"{requested_grantor}\" does not exist"),
277            });
278        }
279        if requested_grantor != current_user {
280            return Err(SQLError::Routine {
281                sqlstate: "0A000".into(),
282                message: "grantor must be current user".into(),
283            });
284        }
285    }
286    Ok(())
287}
288
289pub fn table_sequence_privileges(
290    privileges: &[crate::ast::TablePrivilegeSpec],
291) -> (Vec<SequencePrivilege>, bool) {
292    if privileges.is_empty() {
293        return (
294            vec![
295                SequencePrivilege::Select,
296                SequencePrivilege::Update,
297                SequencePrivilege::Usage,
298            ],
299            false,
300        );
301    }
302    let mut mapped = Vec::new();
303    let mut inapplicable = false;
304    for spec in privileges {
305        let privilege = if spec.columns.is_empty() {
306            match &spec.privilege {
307                TablePrivilege::Select => Some(SequencePrivilege::Select),
308                TablePrivilege::Update => Some(SequencePrivilege::Update),
309                TablePrivilege::Usage => Some(SequencePrivilege::Usage),
310                _ => {
311                    inapplicable = true;
312                    None
313                }
314            }
315        } else {
316            Some(SequencePrivilege::ColumnsUnsupported)
317        };
318        if let Some(privilege) = privilege {
319            if !mapped.contains(&privilege) {
320                mapped.push(privilege);
321            }
322        }
323    }
324    (mapped, inapplicable)
325}
326
327fn table_acl_warning(is_grant: bool, partial: bool, name: &str) -> (&'static str, String) {
328    let message = match (is_grant, partial) {
329        (true, true) => format!("not all privileges were granted for \"{name}\""),
330        (true, false) => format!("no privileges were granted for \"{name}\""),
331        (false, true) => format!("not all privileges could be revoked for \"{name}\""),
332        (false, false) => format!("no privileges could be revoked for \"{name}\""),
333    };
334    ("WARNING", message)
335}
336pub fn view_privilege_updates(
337    targets: Vec<(&ResolvedTableGrantTarget, StoredView)>,
338    application: &TableGrantApplication<'_>,
339    notices: &mut Vec<(&'static str, String)>,
340) -> Result<Vec<ViewPrivilegeUpdate>, SQLError> {
341    let mut updates = Vec::new();
342    for (target, mut view) in targets {
343        let current = view.security();
344        let (next, grantable) = application.apply(&current)?;
345        let columns = view.output_columns.as_deref().ok_or_else(|| {
346            SQLError::Internal(format!(
347                "loaded view `{}` has no durable public column metadata",
348                target.relation.qualified_name()
349            ))
350        })?;
351        validate_table_security_invariants(&next, Some(columns), application.roles).map_err(
352            |error| {
353                SQLError::Internal(format!(
354                    "view `{}` produced invalid privilege metadata: {error}",
355                    target.relation.qualified_name()
356                ))
357            },
358        )?;
359        application.record_warning(grantable, &target.relation, notices);
360        if next != current {
361            view.set_security(next);
362            updates.push((target.relation.clone(), view));
363        }
364    }
365    Ok(updates)
366}
367pub fn foreign_table_privilege_updates(
368    targets: Vec<ForeignTableGrantTarget<'_>>,
369    application: &TableGrantApplication<'_>,
370    notices: &mut Vec<(&'static str, String)>,
371) -> Result<Vec<ForeignTablePrivilegeUpdate>, SQLError> {
372    let mut updates = Vec::new();
373    for (target, current, columns) in targets {
374        let (next, grantable) = application.apply(&current)?;
375        validate_table_security_invariants(&next, Some(&columns), application.roles).map_err(
376            |error| {
377                SQLError::Internal(format!(
378                    "foreign table `{}` produced invalid privilege metadata: {error}",
379                    target.relation.qualified_name()
380                ))
381            },
382        )?;
383        application.record_warning(grantable, &target.relation, notices);
384        if next != current {
385            updates.push((target.relation.clone(), next));
386        }
387    }
388    Ok(updates)
389}
uqa_sql/catalog/security/table_grants.rs

uqa_sql/catalog/security/
table_grants.rs
