uqa_sql/catalog/security/
table.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Table-shaped relation ACL privilege sets, grant paths, and dependency-aware revocation.
8
9use crate::catalog::roles::identity::RoleSubject;
10use std::collections::{BTreeMap, BTreeSet};
11use uqa_core::catalog_acl::AclGrantee;
12
13use super::{TableAclEntry, TablePrivileges};
14use crate::ast::{RoleAttribute, TablePrivilege, TablePrivilegeSpec};
15use crate::SQLError;
16
17use super::TableSecurity;
18use crate::catalog::roles::{role_inherits, RoleDefinition, RoleMembership, RoleMembershipKey};
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
21pub enum TableAclPrivilege {
22    Select,
23    Insert,
24    Update,
25    Delete,
26    Truncate,
27    References,
28    Trigger,
29    Maintain,
30}
31
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33pub struct TablePrivilegeCheck {
34    pub privilege: TableAclPrivilege,
35    pub grant_option: bool,
36}
37
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub struct RequestedTablePrivileges {
40    pub table: Vec<TableAclPrivilege>,
41    pub columns: Vec<(TableAclPrivilege, String)>,
42}
43
44impl TableAclPrivilege {
45    pub const ALL: [Self; 8] = [
46        Self::Select,
47        Self::Insert,
48        Self::Update,
49        Self::Delete,
50        Self::Truncate,
51        Self::References,
52        Self::Trigger,
53        Self::Maintain,
54    ];
55
56    pub const COLUMN_ALL: [Self; 4] = [Self::Select, Self::Insert, Self::Update, Self::References];
57
58    pub const fn mask(self) -> TablePrivileges {
59        let mut privileges = TablePrivileges {
60            select: false,
61            insert: false,
62            update: false,
63            delete: false,
64            truncate: false,
65            references: false,
66            trigger: false,
67            maintain: false,
68        };
69        match self {
70            Self::Select => privileges.select = true,
71            Self::Insert => privileges.insert = true,
72            Self::Update => privileges.update = true,
73            Self::Delete => privileges.delete = true,
74            Self::Truncate => privileges.truncate = true,
75            Self::References => privileges.references = true,
76            Self::Trigger => privileges.trigger = true,
77            Self::Maintain => privileges.maintain = true,
78        }
79        privileges
80    }
81}
82
83pub fn requested_acl_privileges(
84    requested: &[TablePrivilegeSpec],
85) -> Result<RequestedTablePrivileges, SQLError> {
86    if requested.is_empty() {
87        return Ok(RequestedTablePrivileges {
88            table: TableAclPrivilege::ALL.into(),
89            columns: Vec::new(),
90        });
91    }
92    let mut table = Vec::with_capacity(requested.len());
93    let mut columns = Vec::new();
94    for spec in requested {
95        let privilege = match &spec.privilege {
96            TablePrivilege::Select => TableAclPrivilege::Select,
97            TablePrivilege::Insert => TableAclPrivilege::Insert,
98            TablePrivilege::Update => TableAclPrivilege::Update,
99            TablePrivilege::Delete => TableAclPrivilege::Delete,
100            TablePrivilege::Truncate => TableAclPrivilege::Truncate,
101            TablePrivilege::References => TableAclPrivilege::References,
102            TablePrivilege::Trigger => TableAclPrivilege::Trigger,
103            TablePrivilege::Maintain => TableAclPrivilege::Maintain,
104            TablePrivilege::Usage => {
105                return Err(SQLError::Routine {
106                    sqlstate: "0LP01".into(),
107                    message: if spec.columns.is_empty() {
108                        "invalid privilege type USAGE for table".into()
109                    } else {
110                        "invalid privilege type USAGE for column".into()
111                    },
112                })
113            }
114            TablePrivilege::Unsupported(name) => {
115                return Err(SQLError::Routine {
116                    sqlstate: "0LP01".into(),
117                    message: format!(
118                        "invalid privilege type {name} for {}",
119                        if spec.columns.is_empty() {
120                            "table"
121                        } else {
122                            "column"
123                        }
124                    ),
125                })
126            }
127        };
128        if spec.columns.is_empty() {
129            if !table.contains(&privilege) {
130                table.push(privilege);
131            }
132        } else {
133            if !TableAclPrivilege::COLUMN_ALL.contains(&privilege) {
134                return Err(SQLError::Routine {
135                    sqlstate: "0LP01".into(),
136                    message: format!(
137                        "invalid privilege type {} for column",
138                        table_privilege_name(&spec.privilege)
139                    ),
140                });
141            }
142            for column in &spec.columns {
143                let requested = (privilege, column.clone());
144                if !columns.contains(&requested) {
145                    columns.push(requested);
146                }
147            }
148        }
149    }
150    Ok(RequestedTablePrivileges { table, columns })
151}
152
153fn table_privilege_name(privilege: &TablePrivilege) -> &str {
154    match privilege {
155        TablePrivilege::Select => "SELECT",
156        TablePrivilege::Insert => "INSERT",
157        TablePrivilege::Update => "UPDATE",
158        TablePrivilege::Delete => "DELETE",
159        TablePrivilege::Truncate => "TRUNCATE",
160        TablePrivilege::References => "REFERENCES",
161        TablePrivilege::Trigger => "TRIGGER",
162        TablePrivilege::Maintain => "MAINTAIN",
163        TablePrivilege::Usage => "USAGE",
164        TablePrivilege::Unsupported(name) => name,
165    }
166}
167
168pub fn parse_privilege_checks(value: &str) -> Result<Vec<TablePrivilegeCheck>, SQLError> {
169    value
170        .split(',')
171        .map(|item| {
172            let item = item.trim();
173            let upper = item.to_ascii_uppercase();
174            let (name, grant_option) = upper
175                .strip_suffix(" WITH GRANT OPTION")
176                .map_or((upper.as_str(), false), |name| (name.trim_end(), true));
177            let privilege = match name {
178                "SELECT" => TableAclPrivilege::Select,
179                "INSERT" => TableAclPrivilege::Insert,
180                "UPDATE" => TableAclPrivilege::Update,
181                "DELETE" => TableAclPrivilege::Delete,
182                "TRUNCATE" => TableAclPrivilege::Truncate,
183                "REFERENCES" => TableAclPrivilege::References,
184                "TRIGGER" => TableAclPrivilege::Trigger,
185                "MAINTAIN" => TableAclPrivilege::Maintain,
186                _ => {
187                    return Err(SQLError::Routine {
188                        sqlstate: "22023".into(),
189                        message: format!("unrecognized privilege type: \"{item}\""),
190                    })
191                }
192            };
193            Ok(TablePrivilegeCheck {
194                privilege,
195                grant_option,
196            })
197        })
198        .collect()
199}
200
201pub fn parse_column_privilege_checks(value: &str) -> Result<Vec<TablePrivilegeCheck>, SQLError> {
202    let checks = parse_privilege_checks(value)?;
203    if let Some(invalid) = checks
204        .iter()
205        .find(|check| !TableAclPrivilege::COLUMN_ALL.contains(&check.privilege))
206    {
207        let item = value
208            .split(',')
209            .find(|item| {
210                let upper = item.trim().to_ascii_uppercase();
211                let name = upper
212                    .strip_suffix(" WITH GRANT OPTION")
213                    .unwrap_or(&upper)
214                    .trim_end();
215                table_privilege_name_from_acl(invalid.privilege) == name
216            })
217            .map_or(value, str::trim);
218        return Err(SQLError::Routine {
219            sqlstate: "22023".into(),
220            message: format!("unrecognized privilege type: \"{item}\""),
221        });
222    }
223    Ok(checks)
224}
225
226fn table_privilege_name_from_acl(privilege: TableAclPrivilege) -> &'static str {
227    match privilege {
228        TableAclPrivilege::Select => "SELECT",
229        TableAclPrivilege::Insert => "INSERT",
230        TableAclPrivilege::Update => "UPDATE",
231        TableAclPrivilege::Delete => "DELETE",
232        TableAclPrivilege::Truncate => "TRUNCATE",
233        TableAclPrivilege::References => "REFERENCES",
234        TableAclPrivilege::Trigger => "TRIGGER",
235        TableAclPrivilege::Maintain => "MAINTAIN",
236    }
237}
238
239pub fn acl_grantor<'a>(entry: &'a TableAclEntry, owner: &'a str) -> &'a str {
240    entry.grantor.as_deref().unwrap_or(owner)
241}
242
243fn materialize_acl(security: &mut TableSecurity) {
244    if security.acl.is_none() {
245        security.acl = Some(vec![TableAclEntry {
246            role: security.role_owner.clone().into(),
247            grantor: Some(security.role_owner.clone()),
248            privileges: TablePrivileges::ALL,
249            grant_options: TablePrivileges::default(),
250        }]);
251    }
252}
253
254pub fn grant_option_roles(
255    security: &TableSecurity,
256    privilege: TableAclPrivilege,
257) -> BTreeSet<String> {
258    let mut reachable = BTreeSet::from([security.role_owner.clone()]);
259    let Some(acl) = security.acl.as_ref() else {
260        return reachable;
261    };
262    loop {
263        let mut changed = false;
264        for entry in acl {
265            let Some(role) = entry.role.role_name() else {
266                continue;
267            };
268            if entry.grant_options.intersects(privilege.mask())
269                && reachable.contains(acl_grantor(entry, &security.role_owner))
270            {
271                changed |= reachable.insert(role.to_owned());
272            }
273        }
274        if !changed {
275            return reachable;
276        }
277    }
278}
279
280pub fn select_acl_grantor(
281    security: &TableSecurity,
282    privilege: TableAclPrivilege,
283    current_user: &(impl RoleSubject + ?Sized),
284    roles: &BTreeMap<String, RoleDefinition>,
285    memberships: &BTreeMap<RoleMembershipKey, RoleMembership>,
286) -> Option<String> {
287    let current_user = current_user.role_name(roles)?;
288    if role_inherits(roles, memberships, current_user, &security.role_owner) {
289        return Some(security.role_owner.clone());
290    }
291    let grant_options = grant_option_roles(security, privilege);
292    if grant_options.contains(current_user) {
293        return Some(current_user.to_string());
294    }
295    security.acl.as_ref().and_then(|acl| {
296        acl.iter()
297            .filter_map(|entry| entry.role.role_name())
298            .filter(|role| grant_options.contains(*role))
299            .find(|role| role_inherits(roles, memberships, current_user, *role))
300            .map(str::to_owned)
301    })
302}
303
304pub fn role_has_privilege(
305    security: &TableSecurity,
306    subject: &(impl RoleSubject + ?Sized),
307    check: TablePrivilegeCheck,
308    roles: &BTreeMap<String, RoleDefinition>,
309    memberships: &BTreeMap<RoleMembershipKey, RoleMembership>,
310) -> bool {
311    if subject
312        .role_definition(roles)
313        .is_some_and(|role| role.has(RoleAttribute::Superuser))
314    {
315        return true;
316    }
317    if check.grant_option {
318        return grant_option_roles(security, check.privilege)
319            .iter()
320            .any(|role| role_inherits(roles, memberships, subject, role));
321    }
322    match security.acl.as_ref() {
323        None => role_inherits(roles, memberships, subject, &security.role_owner),
324        Some(acl) => acl.iter().any(|entry| {
325            entry.privileges.intersects(check.privilege.mask())
326                && (entry.role.is_public()
327                    || role_inherits(roles, memberships, subject, &entry.role))
328        }),
329    }
330}
331
332pub fn grant_acl(
333    security: &mut TableSecurity,
334    privilege: TableAclPrivilege,
335    grantees: &[AclGrantee],
336    grantor: &str,
337    grant_option: bool,
338) {
339    materialize_acl(security);
340    let owner = security.role_owner.clone();
341    let acl = security.acl.as_mut().expect("table ACL was materialized");
342    for grantee in grantees {
343        let position = acl
344            .iter()
345            .position(|entry| entry.role == *grantee && acl_grantor(entry, &owner) == grantor)
346            .unwrap_or_else(|| {
347                acl.push(TableAclEntry {
348                    role: grantee.clone(),
349                    grantor: Some(grantor.to_string()),
350                    privileges: TablePrivileges::default(),
351                    grant_options: TablePrivileges::default(),
352                });
353                acl.len() - 1
354            });
355        let entry = &mut acl[position];
356        entry.privileges.insert(privilege.mask());
357        if grant_option && grantee.role_name().is_some_and(|name| name != owner) {
358            entry.grant_options.insert(privilege.mask());
359        }
360    }
361}
362
363pub fn revoke_acl(
364    security: &mut TableSecurity,
365    privilege: TableAclPrivilege,
366    grantees: &[AclGrantee],
367    grantor: &str,
368    grant_option_only: bool,
369    cascade: bool,
370) -> Result<(), SQLError> {
371    let before = grant_option_roles(security, privilege);
372    materialize_acl(security);
373    let owner = security.role_owner.clone();
374    let acl = security.acl.as_mut().expect("table ACL was materialized");
375    for entry in acl
376        .iter_mut()
377        .filter(|entry| grantees.contains(&entry.role) && acl_grantor(entry, &owner) == grantor)
378    {
379        entry.grant_options.remove(privilege.mask());
380        if !grant_option_only {
381            entry.privileges.remove(privilege.mask());
382        }
383    }
384    remove_empty_entries(acl);
385    revoke_dependent_acl(security, privilege, &before, cascade)
386}
387
388fn revoke_dependent_acl(
389    security: &mut TableSecurity,
390    privilege: TableAclPrivilege,
391    before: &BTreeSet<String>,
392    cascade: bool,
393) -> Result<(), SQLError> {
394    loop {
395        let current = grant_option_roles(security, privilege);
396        let lost = before
397            .difference(&current)
398            .cloned()
399            .collect::<BTreeSet<_>>();
400        if lost.is_empty() {
401            return Ok(());
402        }
403        let owner = security.role_owner.clone();
404        let dependent = security.acl.as_ref().is_some_and(|acl| {
405            acl.iter().any(|entry| {
406                lost.contains(acl_grantor(entry, &owner))
407                    && (entry.privileges.intersects(privilege.mask())
408                        || entry.grant_options.intersects(privilege.mask()))
409            })
410        });
411        if !dependent {
412            return Ok(());
413        }
414        if !cascade {
415            return Err(SQLError::Routine {
416                sqlstate: "2BP01".into(),
417                message: "dependent privileges exist".into(),
418            });
419        }
420        let acl = security
421            .acl
422            .as_mut()
423            .expect("dependent table privileges require an explicit ACL");
424        for entry in acl
425            .iter_mut()
426            .filter(|entry| lost.contains(acl_grantor(entry, &owner)))
427        {
428            entry.privileges.remove(privilege.mask());
429            entry.grant_options.remove(privilege.mask());
430        }
431        remove_empty_entries(acl);
432    }
433}
434
435fn remove_empty_entries(acl: &mut Vec<TableAclEntry>) {
436    acl.retain(|entry| !entry.privileges.is_empty() || !entry.grant_options.is_empty());
437}
438
439pub fn rewrite_acl_owner(security: &mut TableSecurity, new_owner: &str) {
440    let old_owner = std::mem::replace(&mut security.role_owner, new_owner.to_string());
441    if let Some(acl) = security.acl.as_mut() {
442        rewrite_acl_entries_owner(acl, &old_owner, new_owner);
443    }
444    for column_acl in security.column_acls.values_mut() {
445        rewrite_acl_entries_owner(column_acl, &old_owner, new_owner);
446    }
447}
448
449fn rewrite_acl_entries_owner(acl: &mut Vec<TableAclEntry>, old_owner: &str, new_owner: &str) {
450    for entry in acl.iter_mut() {
451        if entry.role.role_name() == Some(old_owner) {
452            entry.role = new_owner.into();
453        }
454        if entry.grantor.as_deref() == Some(old_owner) {
455            entry.grantor = Some(new_owner.to_string());
456        }
457    }
458    let mut merged: Vec<TableAclEntry> = Vec::with_capacity(acl.len());
459    for entry in std::mem::take(acl) {
460        if let Some(existing) = merged.iter_mut().find(|existing| {
461            existing.role == entry.role
462                && acl_grantor(existing, new_owner) == acl_grantor(&entry, new_owner)
463        }) {
464            existing.privileges.insert(entry.privileges);
465            existing.grant_options.insert(entry.grant_options);
466        } else {
467            merged.push(entry);
468        }
469    }
470    *acl = merged;
471}
472
473pub fn role_can_view_table(
474    security: &TableSecurity,
475    subject: &(impl RoleSubject + ?Sized),
476    roles: &BTreeMap<String, RoleDefinition>,
477    memberships: &BTreeMap<RoleMembershipKey, RoleMembership>,
478) -> bool {
479    TableAclPrivilege::ALL.into_iter().any(|privilege| {
480        role_has_privilege(
481            security,
482            subject,
483            TablePrivilegeCheck {
484                privilege,
485                grant_option: false,
486            },
487            roles,
488            memberships,
489        )
490    }) || security.column_acls.iter().any(|(column, _)| {
491        TableAclPrivilege::COLUMN_ALL.into_iter().any(|privilege| {
492            super::columns::role_has_column_privilege(
493                security,
494                column,
495                subject,
496                TablePrivilegeCheck {
497                    privilege,
498                    grant_option: false,
499                },
500                roles,
501                memberships,
502            )
503        })
504    })
505}
506
507pub fn role_has_table_privilege(
508    security: &TableSecurity,
509    subject: &(impl RoleSubject + ?Sized),
510    privilege: TableAclPrivilege,
511    roles: &BTreeMap<String, RoleDefinition>,
512    memberships: &BTreeMap<RoleMembershipKey, RoleMembership>,
513) -> bool {
514    role_has_privilege(
515        security,
516        subject,
517        TablePrivilegeCheck {
518            privilege,
519            grant_option: false,
520        },
521        roles,
522        memberships,
523    )
524}
525
526pub fn role_has_column_privilege(
527    security: &TableSecurity,
528    column: &str,
529    subject: &(impl RoleSubject + ?Sized),
530    privilege: TableAclPrivilege,
531    roles: &BTreeMap<String, RoleDefinition>,
532    memberships: &BTreeMap<RoleMembershipKey, RoleMembership>,
533) -> bool {
534    super::columns::role_has_column_privilege(
535        security,
536        column,
537        subject,
538        TablePrivilegeCheck {
539            privilege,
540            grant_option: false,
541        },
542        roles,
543        memberships,
544    )
545}
546
547mod invariants;
548pub use invariants::validate_table_security_invariants;
uqa_sql/catalog/security/table.rs

uqa_sql/catalog/security/
table.rs
