uqa_sql/catalog/security/
schema_inquiry.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Schema privilege inquiry, namespace visibility and default security rules.
8
9use super::{
10    schema::{
11        parse_privilege_checks, role_has_schema_privilege, role_has_schema_privilege_check,
12        SchemaAclPrivilege,
13    },
14    BoundSchemaSecurity,
15};
16use crate::catalog::roles::identity::RoleSubject;
17use crate::catalog::roles::RoleReference;
18use crate::{
19    catalog::roles::{guards::RoleCatalogGuards, RoleDefinition, RoleReferenceNames},
20    SQLError,
21};
22use std::collections::{BTreeMap, BTreeSet};
23use uqa_core::Value;
24
25pub type SchemaRegistryRead<'a> =
26    Box<dyn std::ops::Deref<Target = BTreeMap<String, BoundSchemaSecurity>> + 'a>;
27
28/// Metadata-only graph names held under the caller's original registry read guard.
29pub trait GraphNamespaceRead {
30    fn names(&self) -> Box<dyn Iterator<Item = &str> + '_>;
31    fn contains(&self, name: &str) -> bool;
32}
33
34pub trait SchemaPrivilegeCatalog {
35    fn refresh_namespace_catalog(&self) -> Result<(), SQLError>;
36    fn schemas(&self) -> SchemaRegistryRead<'_>;
37    fn graphs(&self) -> Box<dyn GraphNamespaceRead + '_>;
38    fn temporary_namespace_allocated(&self) -> bool;
39    fn temporary_schema_name(&self) -> String;
40}
41
42pub struct SchemaPrivilegeInquiry<'a> {
43    pub catalog: &'a dyn SchemaPrivilegeCatalog,
44    pub names: &'a dyn RoleReferenceNames,
45    pub roles: &'a dyn RoleCatalogGuards,
46}
47
48impl SchemaPrivilegeInquiry<'_> {
49    pub fn schema_has_privilege_for_role(
50        &self,
51        schema: &str,
52        role: &(impl RoleSubject + ?Sized),
53        privilege: SchemaAclPrivilege,
54    ) -> bool {
55        let Some(security) = self.schema_security_for_privilege(schema) else {
56            return false;
57        };
58        let roles = self.roles.role_definitions();
59        let memberships = self.roles.role_memberships();
60        security.resolve(&roles).is_ok_and(|security| {
61            role_has_schema_privilege(&security, role, privilege, &roles, &memberships)
62        })
63    }
64
65    pub fn schema_security_for_privilege(&self, schema: &str) -> Option<BoundSchemaSecurity> {
66        if let Some(security) = self.catalog.schemas().get(schema) {
67            return Some(security.clone());
68        }
69        let mut security = match schema {
70            "pg_catalog" | "information_schema" => {
71                Some(BoundSchemaSecurity::with_public_privileges(false))
72            }
73            "ag_catalog" => Some(BoundSchemaSecurity::bootstrap("ag_catalog")),
74            name if name == self.catalog.temporary_schema_name() => {
75                Some(BoundSchemaSecurity::with_public_privileges(true))
76            }
77            name if self.catalog.graphs().contains(name) => {
78                Some(BoundSchemaSecurity::bootstrap(name))
79            }
80            _ => None,
81        }?;
82        security.tuple = BoundSchemaSecurity::bootstrap(schema).tuple;
83        Some(security)
84    }
85
86    pub fn require_schema_privilege(
87        &self,
88        schema: &str,
89        role: &(impl RoleSubject + ?Sized),
90        privilege: SchemaAclPrivilege,
91    ) -> Result<(), SQLError> {
92        if self.schema_has_privilege_for_role(schema, role, privilege) {
93            return Ok(());
94        }
95        Err(SQLError::Routine {
96            sqlstate: "42501".into(),
97            message: format!("permission denied for schema {schema}"),
98        })
99    }
100
101    pub fn has_schema_privilege_value(&self, arguments: &[Value]) -> Result<Value, SQLError> {
102        if arguments.iter().any(|argument| argument == &Value::Null) {
103            return Ok(Value::Null);
104        }
105        let (subject_value, schema_value, privilege_value) = match arguments {
106            [schema, privilege] => (None, schema, privilege),
107            [subject, schema, privilege] => (Some(subject), schema, privilege),
108            _ => {
109                return Err(SQLError::BadArity {
110                    name: "has_schema_privilege".into(),
111                    expected: "2 or 3".into(),
112                    actual: arguments.len(),
113                })
114            }
115        };
116        let current_user = subject_value.is_none().then(|| self.names.current_role());
117        let subject = {
118            let roles = self.roles.role_definitions();
119            subject_value.map_or_else(
120                || Ok(current_user),
121                |value| {
122                    resolve_schema_privilege_role(value, &roles)
123                        .map(|role| role.map(RoleReference::from))
124                },
125            )?
126        };
127        let schema = self.resolve_schema_privilege_target(schema_value)?;
128        let privilege = match privilege_value {
129            Value::Str(privilege) | Value::FixedChar(privilege) => privilege,
130            other => {
131                return Err(SQLError::TypeMismatch(format!(
132                    "has_schema_privilege privilege must be text, got {other:?}"
133                )))
134            }
135        };
136        let checks = parse_privilege_checks(privilege)?;
137        let roles = self.roles.role_definitions();
138        let memberships = self.roles.role_memberships();
139        let subject_is_superuser = subject.as_ref().is_some_and(|subject| {
140            subject
141                .role_definition(&roles)
142                .is_some_and(|role| role.has(crate::ast::RoleAttribute::Superuser))
143        });
144        let Some(schema) = schema else {
145            return if subject_is_superuser {
146                Ok(Value::Bool(true))
147            } else {
148                Ok(Value::Null)
149            };
150        };
151        let Some(subject) = subject else {
152            return Ok(Value::Bool(false));
153        };
154        let security = self.schema_security_for_privilege(&schema).ok_or_else(|| {
155            SQLError::Internal(format!("schema `{schema}` has no security metadata"))
156        })?;
157        let security = security.resolve(&roles).map_err(SQLError::Internal)?;
158        Ok(Value::Bool(checks.into_iter().any(|check| {
159            role_has_schema_privilege_check(&security, &subject, check, &roles, &memberships)
160        })))
161    }
162
163    fn resolve_schema_privilege_target(&self, value: &Value) -> Result<Option<String>, SQLError> {
164        let names = self.schema_privilege_namespace_names()?;
165        match value {
166            Value::Str(name) | Value::FixedChar(name) => {
167                if names.contains(name) {
168                    Ok(Some(name.clone()))
169                } else {
170                    Err(SQLError::Routine {
171                        sqlstate: "3F000".into(),
172                        message: format!("schema \"{name}\" does not exist"),
173                    })
174                }
175            }
176            Value::Int(oid) => Ok(names.into_iter().find(|name| {
177                self.schema_security_for_privilege(name)
178                    .is_some_and(|security| security.namespace_oid(name) == *oid)
179            })),
180            other => Err(SQLError::TypeMismatch(format!(
181                "has_schema_privilege schema must be text or oid, got {other:?}"
182            ))),
183        }
184    }
185
186    fn schema_privilege_namespace_names(&self) -> Result<BTreeSet<String>, SQLError> {
187        self.catalog.refresh_namespace_catalog()?;
188        let mut names = BTreeSet::from([
189            "pg_catalog".to_string(),
190            "information_schema".to_string(),
191            "ag_catalog".to_string(),
192        ]);
193        names.extend(self.catalog.schemas().keys().cloned());
194        names.extend(self.catalog.graphs().names().map(str::to_owned));
195        if self.catalog.temporary_namespace_allocated() {
196            names.insert(self.catalog.temporary_schema_name());
197        }
198        Ok(names)
199    }
200}
201
202fn resolve_schema_privilege_role(
203    value: &Value,
204    roles: &BTreeMap<String, RoleDefinition>,
205) -> Result<Option<String>, SQLError> {
206    match value {
207        Value::Str(name) | Value::FixedChar(name) => {
208            if roles.contains_key(name) {
209                Ok(Some(name.clone()))
210            } else {
211                Err(SQLError::Routine {
212                    sqlstate: "42704".into(),
213                    message: format!("role \"{name}\" does not exist"),
214                })
215            }
216        }
217        Value::Int(oid) => Ok(roles
218            .values()
219            .find(|role| role.oid == *oid)
220            .map(|role| role.name.clone())),
221        other => Err(SQLError::TypeMismatch(format!(
222            "has_schema_privilege role must be name or oid, got {other:?}"
223        ))),
224    }
225}
uqa_sql/catalog/security/schema_inquiry.rs

uqa_sql/catalog/security/
schema_inquiry.rs
