uqa_sql/catalog/security/
schema.rs

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

uqa_sql/catalog/security/
schema.rs
