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

uqa_sql/catalog/security/
table.rs
