uqa_sql/catalog/security/
database.rs

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

uqa_sql/catalog/security/
database.rs
