uqa-execution 0.3.0

Volcano physical operators with row-batch pipelines
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Database ACL publication and restoration with retained authorization guards.

use std::ops::DerefMut;
use uqa_sql::{
    ast::GrantDatabaseStmt,
    catalog::{
        roles::{guards::RoleCatalogGuards, resolve_role_reference, RoleReferenceNames},
        security::{
            database::{
                apply_database_acl, database_acl_warning, requested_acl_privileges,
                resolve_database_grant_targets, validate_database_acl_roles,
                validate_stored_database_security, DatabaseSecurity,
            },
            database_inquiry::DatabaseSecurityRead,
        },
        DATABASE_NAME,
    },
    SQLError,
};
use uqa_storage::{CatalogFacade, StorageBackendError, StorageBackendResult};

pub const DATABASE_SECURITY_METADATA_KEY: &str = "sql_database_security_json";
pub type DatabaseSecurityWrite<'a> = Box<dyn DerefMut<Target = DatabaseSecurity> + 'a>;

pub trait DatabaseSecurityRegistry {
    fn security_read(&self) -> DatabaseSecurityRead<'_>;
    fn security_write(&self) -> DatabaseSecurityWrite<'_>;
}
pub trait DatabasePrivilegePublication {
    fn prepare_writer(&self) -> Result<(), SQLError>;
    fn refresh_catalog(&self) -> StorageBackendResult<()>;
    fn persist_security(&self, security: &DatabaseSecurity) -> Result<(), SQLError>;
    fn catalog_changed(&self);
    fn notice(&self, level: &str, message: &str);
}
pub struct DatabasePrivilegeContext<'a> {
    pub names: &'a dyn RoleReferenceNames,
    pub roles: &'a dyn RoleCatalogGuards,
    pub registry: &'a dyn DatabaseSecurityRegistry,
    pub publication: &'a dyn DatabasePrivilegePublication,
}

pub fn grant_database_privileges(
    context: &DatabasePrivilegeContext<'_>,
    statement: &GrantDatabaseStmt,
) -> Result<(), SQLError> {
    context.publication.prepare_writer()?;
    context
        .publication
        .refresh_catalog()
        .map_err(|error| SQLError::Internal(format!("load database privileges: {error}")))?;
    resolve_database_grant_targets(&statement.databases)?;
    let grantees = statement
        .grantees
        .iter()
        .map(|role| resolve_role_reference(context.names, role))
        .collect::<Vec<_>>();
    let requested_grantor = statement
        .grantor
        .as_ref()
        .map(|role| resolve_role_reference(context.names, role));
    let current_user = context.names.current_user_name();
    let roles = context.roles.role_definitions();
    validate_database_acl_roles(
        statement,
        &grantees,
        requested_grantor.as_deref(),
        &current_user,
        &roles,
    )?;
    let privileges = requested_acl_privileges(&statement.privileges)?;
    let memberships = context.roles.role_memberships();
    let current = context.registry.security_read().clone();
    let (next, grantable) = apply_database_acl(
        statement,
        &grantees,
        &privileges,
        &current_user,
        &roles,
        &memberships,
        &current,
    )?;
    let notice = (grantable != privileges.len())
        .then(|| database_acl_warning(statement.is_grant, grantable != 0, DATABASE_NAME));
    if next != current {
        context.publication.persist_security(&next)?;
        **context.registry.security_write() = next;
        context.publication.catalog_changed();
    }
    drop(memberships);
    drop(roles);
    if let Some((level, message)) = notice {
        context.publication.notice(level, &message);
    }
    Ok(())
}

pub fn restore_database_security_from_metadata(
    context: &DatabasePrivilegeContext<'_>,
    catalog: &dyn CatalogFacade,
) -> StorageBackendResult<()> {
    let security = match catalog.get_metadata(DATABASE_SECURITY_METADATA_KEY)? {
        Some(json) => serde_json::from_str::<DatabaseSecurity>(&json)?,
        None => DatabaseSecurity::bootstrap(),
    };
    let roles = context.roles.role_definitions();
    validate_stored_database_security(&security, &roles).map_err(StorageBackendError::Other)?;
    drop(roles);
    **context.registry.security_write() = security;
    Ok(())
}

#[cfg(test)]
mod tests;