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

uqa_sql/catalog/security/
schema_inquiry.rs
