uqa_sql/catalog/security/
database_inquiry.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Database privilege inquiry with ordered identity, catalog and security reads.
8
9use super::database::{
10    parse_privilege_checks, role_has_database_privilege, role_has_database_privilege_check,
11    BoundDatabaseSecurity, DatabaseAclPrivilege,
12};
13use crate::catalog::roles::identity::RoleSubject;
14use crate::catalog::roles::RoleReference;
15use crate::{
16    catalog::{
17        roles::{guards::RoleCatalogGuards, RoleDefinition, RoleReferenceNames},
18        DATABASE_NAME, DATABASE_OID,
19    },
20    SQLError,
21};
22use std::collections::BTreeMap;
23use uqa_core::Value;
24
25pub type DatabaseSecurityRead<'a> = Box<dyn std::ops::Deref<Target = BoundDatabaseSecurity> + 'a>;
26
27pub trait DatabasePrivilegeCatalog {
28    fn refresh_privilege_catalog(&self) -> Result<(), SQLError>;
29    fn security(&self) -> DatabaseSecurityRead<'_>;
30}
31
32pub struct DatabasePrivilegeInquiry<'a> {
33    pub catalog: &'a dyn DatabasePrivilegeCatalog,
34    pub names: &'a dyn RoleReferenceNames,
35    pub roles: &'a dyn RoleCatalogGuards,
36}
37
38impl DatabasePrivilegeInquiry<'_> {
39    pub fn ensure_database_privilege(
40        &self,
41        role: &(impl RoleSubject + ?Sized),
42        privilege: DatabaseAclPrivilege,
43    ) -> Result<(), SQLError> {
44        let security = self.catalog.security();
45        let roles = self.roles.role_definitions();
46        let memberships = self.roles.role_memberships();
47        let resolved = security.resolve(&roles).map_err(SQLError::Internal)?;
48        if role_has_database_privilege(&resolved, role, privilege, &roles, &memberships) {
49            return Ok(());
50        }
51        let message = match privilege {
52            DatabaseAclPrivilege::Temporary => {
53                format!(
54                    "permission denied to create temporary tables in database \"{DATABASE_NAME}\""
55                )
56            }
57            DatabaseAclPrivilege::Connect | DatabaseAclPrivilege::Create => {
58                format!("permission denied for database {DATABASE_NAME}")
59            }
60        };
61        Err(SQLError::Routine {
62            sqlstate: "42501".into(),
63            message,
64        })
65    }
66
67    pub fn has_database_privilege_value(&self, arguments: &[Value]) -> Result<Value, SQLError> {
68        if arguments.iter().any(|argument| argument == &Value::Null) {
69            return Ok(Value::Null);
70        }
71        self.catalog.refresh_privilege_catalog()?;
72        let (subject_value, database_value, privilege_value) = match arguments {
73            [database, privilege] => (None, database, privilege),
74            [subject, database, privilege] => (Some(subject), database, privilege),
75            _ => {
76                return Err(SQLError::BadArity {
77                    name: "has_database_privilege".into(),
78                    expected: "2 or 3".into(),
79                    actual: arguments.len(),
80                })
81            }
82        };
83        let current_user = subject_value.is_none().then(|| self.names.current_role());
84        let subject = {
85            let roles = self.roles.role_definitions();
86            subject_value.map_or_else(
87                || Ok(current_user),
88                |value| {
89                    resolve_database_privilege_role(value, &roles)
90                        .map(|role| role.map(RoleReference::from))
91                },
92            )?
93        };
94        let database_exists = resolve_database_privilege_target(database_value)?;
95        let privilege = match privilege_value {
96            Value::Str(privilege) | Value::FixedChar(privilege) => privilege,
97            other => {
98                return Err(SQLError::TypeMismatch(format!(
99                    "has_database_privilege privilege must be text, got {other:?}"
100                )))
101            }
102        };
103        let checks = parse_privilege_checks(privilege)?;
104        let roles = self.roles.role_definitions();
105        let memberships = self.roles.role_memberships();
106        let subject_is_superuser = subject.as_ref().is_some_and(|subject| {
107            subject
108                .role_definition(&roles)
109                .is_some_and(|role| role.has(crate::ast::RoleAttribute::Superuser))
110        });
111        if !database_exists {
112            return if subject_is_superuser {
113                Ok(Value::Bool(true))
114            } else {
115                Ok(Value::Null)
116            };
117        }
118        let Some(subject) = subject else {
119            return Ok(Value::Bool(false));
120        };
121        let security = self.catalog.security();
122        let resolved = security.resolve(&roles).map_err(SQLError::Internal)?;
123        Ok(Value::Bool(checks.into_iter().any(|check| {
124            role_has_database_privilege_check(&resolved, &subject, check, &roles, &memberships)
125        })))
126    }
127}
128
129fn resolve_database_privilege_target(value: &Value) -> Result<bool, SQLError> {
130    match value {
131        Value::Str(name) | Value::FixedChar(name) => {
132            if name == DATABASE_NAME {
133                Ok(true)
134            } else {
135                Err(SQLError::Routine {
136                    sqlstate: "3D000".into(),
137                    message: format!("database \"{name}\" does not exist"),
138                })
139            }
140        }
141        Value::Int(oid) => Ok(*oid == DATABASE_OID),
142        other => Err(SQLError::TypeMismatch(format!(
143            "has_database_privilege database must be text or oid, got {other:?}"
144        ))),
145    }
146}
147
148fn resolve_database_privilege_role(
149    value: &Value,
150    roles: &BTreeMap<String, RoleDefinition>,
151) -> Result<Option<String>, SQLError> {
152    match value {
153        Value::Str(name) | Value::FixedChar(name) => {
154            if roles.contains_key(name) {
155                Ok(Some(name.clone()))
156            } else {
157                Err(SQLError::Routine {
158                    sqlstate: "42704".into(),
159                    message: format!("role \"{name}\" does not exist"),
160                })
161            }
162        }
163        Value::Int(oid) => Ok(roles
164            .values()
165            .find(|role| role.oid == *oid)
166            .map(|role| role.name.clone())),
167        other => Err(SQLError::TypeMismatch(format!(
168            "has_database_privilege role must be name or oid, got {other:?}"
169        ))),
170    }
171}
uqa_sql/catalog/security/database_inquiry.rs

uqa_sql/catalog/security/
database_inquiry.rs
