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

uqa_sql/catalog/security/
schema_inquiry.rs
