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

uqa_sql/catalog/security/
database.rs
