uqa_sql/catalog/security/
sequence_inquiry.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Sequence privilege inquiry, target binding and access checks.
8
9use super::{
10    sequence::{self as acl, role_has_privilege, AclPrivilege, PrivilegeCheck},
11    SequenceSecurity,
12};
13use crate::{
14    catalog::{
15        resolution::RelationResolution,
16        roles::{guards::RoleCatalogGuards, RoleDefinition, RoleReferenceNames},
17    },
18    SQLError,
19};
20use std::{collections::BTreeMap, ops::Deref};
21use uqa_core::{RelationIdentity, Value};
22
23pub type SequenceSecurityRead<'a> =
24    Box<dyn Deref<Target = BTreeMap<RelationIdentity, SequenceSecurity>> + 'a>;
25pub trait SequenceSecurityCatalog {
26    fn security_read(&self) -> SequenceSecurityRead<'_>;
27}
28pub trait SequencePrivilegeResolution {
29    fn visible_relation_kind(&self, reference: &str) -> Result<RelationResolution, SQLError>;
30    fn sequence_privilege_oid(
31        &self,
32        oid: i64,
33    ) -> Result<Option<(String, RelationIdentity)>, SQLError>;
34}
35pub struct SequencePrivilegeInquiry<'a> {
36    pub names: &'a dyn RoleReferenceNames,
37    pub roles: &'a dyn RoleCatalogGuards,
38    pub security: &'a dyn SequenceSecurityCatalog,
39    pub resolution: &'a dyn SequencePrivilegeResolution,
40}
41
42impl SequencePrivilegeInquiry<'_> {
43    pub fn ensure_sequence_nextval_privilege(
44        &self,
45        name: &str,
46        relation: &RelationIdentity,
47    ) -> Result<(), SQLError> {
48        self.ensure_sequence_value_privilege(
49            name,
50            relation,
51            &[AclPrivilege::Usage, AclPrivilege::Update],
52        )
53    }
54
55    pub fn ensure_sequence_currval_privilege(
56        &self,
57        name: &str,
58        relation: &RelationIdentity,
59    ) -> Result<(), SQLError> {
60        self.ensure_sequence_value_privilege(
61            name,
62            relation,
63            &[AclPrivilege::Usage, AclPrivilege::Select],
64        )
65    }
66
67    pub fn ensure_sequence_setval_privilege(
68        &self,
69        name: &str,
70        relation: &RelationIdentity,
71    ) -> Result<(), SQLError> {
72        self.ensure_sequence_value_privilege(name, relation, &[AclPrivilege::Update])
73    }
74
75    fn ensure_sequence_value_privilege(
76        &self,
77        name: &str,
78        relation: &RelationIdentity,
79        privileges: &[AclPrivilege],
80    ) -> Result<(), SQLError> {
81        let security = self
82            .security
83            .security_read()
84            .get(relation)
85            .cloned()
86            .ok_or_else(|| {
87                SQLError::Internal(format!("sequence `{name}` has no security metadata"))
88            })?;
89        let current_user = self.names.current_user_name();
90        let roles = self.roles.role_definitions();
91        let memberships = self.roles.role_memberships();
92        if privileges.iter().any(|privilege| {
93            role_has_privilege(
94                &security,
95                &current_user,
96                PrivilegeCheck {
97                    privilege: *privilege,
98                    grant_option: false,
99                },
100                &roles,
101                &memberships,
102            )
103        }) {
104            return Ok(());
105        }
106        Err(SQLError::Routine {
107            sqlstate: "42501".into(),
108            message: format!("permission denied for sequence {}", relation.name),
109        })
110    }
111
112    pub fn has_sequence_privilege_value(&self, arguments: &[Value]) -> Result<Value, SQLError> {
113        if arguments.iter().any(|argument| argument == &Value::Null) {
114            return Ok(Value::Null);
115        }
116        let (subject_value, sequence_value, privilege_value) = match arguments {
117            [sequence, privilege] => (None, sequence, privilege),
118            [subject, sequence, privilege] => (Some(subject), sequence, privilege),
119            _ => {
120                return Err(SQLError::BadArity {
121                    name: "has_sequence_privilege".into(),
122                    expected: "2 or 3".into(),
123                    actual: arguments.len(),
124                })
125            }
126        };
127        let current_user = subject_value
128            .is_none()
129            .then(|| self.names.current_user_name());
130        let subject = {
131            let roles = self.roles.role_definitions();
132            subject_value.map_or_else(
133                || Ok(current_user),
134                |value| resolve_sequence_privilege_role(value, &roles),
135            )?
136        };
137        let Some((_name, relation)) = self.resolve_sequence_privilege_target(sequence_value)?
138        else {
139            return Ok(Value::Null);
140        };
141        let privilege = match privilege_value {
142            Value::Str(privilege) | Value::FixedChar(privilege) => privilege,
143            other => {
144                return Err(SQLError::TypeMismatch(format!(
145                    "has_sequence_privilege privilege must be text, got {other:?}"
146                )))
147            }
148        };
149        let checks = acl::parse_privilege_checks(privilege)?;
150        let Some(subject) = subject else {
151            return Ok(Value::Bool(false));
152        };
153        let security = self
154            .security
155            .security_read()
156            .get(&relation)
157            .cloned()
158            .ok_or_else(|| {
159                SQLError::Internal(format!(
160                    "sequence `{}` has no security metadata",
161                    relation.qualified_name()
162                ))
163            })?;
164        let roles = self.roles.role_definitions();
165        let memberships = self.roles.role_memberships();
166        Ok(Value::Bool(checks.into_iter().any(|check| {
167            role_has_privilege(&security, &subject, check, &roles, &memberships)
168        })))
169    }
170
171    pub fn role_has_sequence_table_privilege(
172        &self,
173        relation: &RelationIdentity,
174        subject: &str,
175        privilege: super::table::TableAclPrivilege,
176        grant_option: bool,
177    ) -> Result<bool, SQLError> {
178        let privilege = match privilege {
179            super::table::TableAclPrivilege::Select => AclPrivilege::Select,
180            super::table::TableAclPrivilege::Update => AclPrivilege::Update,
181            super::table::TableAclPrivilege::Insert
182            | super::table::TableAclPrivilege::Delete
183            | super::table::TableAclPrivilege::Truncate
184            | super::table::TableAclPrivilege::References
185            | super::table::TableAclPrivilege::Trigger
186            | super::table::TableAclPrivilege::Maintain => return Ok(false),
187        };
188        let security = self
189            .security
190            .security_read()
191            .get(relation)
192            .cloned()
193            .ok_or_else(|| {
194                SQLError::Internal(format!(
195                    "sequence `{}` has no security metadata",
196                    relation.qualified_name()
197                ))
198            })?;
199        let roles = self.roles.role_definitions();
200        let memberships = self.roles.role_memberships();
201        Ok(role_has_privilege(
202            &security,
203            subject,
204            PrivilegeCheck {
205                privilege,
206                grant_option,
207            },
208            &roles,
209            &memberships,
210        ))
211    }
212
213    fn resolve_sequence_privilege_target(
214        &self,
215        value: &Value,
216    ) -> Result<Option<(String, RelationIdentity)>, SQLError> {
217        match value {
218            Value::Str(reference) | Value::FixedChar(reference) => {
219                let (name, kind) = match self.resolution.visible_relation_kind(reference)? {
220                    RelationResolution::Found(name, kind) => (name, kind),
221                    RelationResolution::MissingSchema(schema) => {
222                        return Err(SQLError::Routine {
223                            sqlstate: "3F000".into(),
224                            message: format!("schema \"{schema}\" does not exist"),
225                        });
226                    }
227                    RelationResolution::MissingRelation => {
228                        return Err(SQLError::Routine {
229                            sqlstate: "42P01".into(),
230                            message: format!("relation \"{reference}\" does not exist"),
231                        });
232                    }
233                };
234                if kind != "sequence" {
235                    return Err(SQLError::Routine {
236                        sqlstate: "42809".into(),
237                        message: format!("\"{reference}\" is not a sequence"),
238                    });
239                }
240                let relation = RelationIdentity::from_legacy_name(&name).map_err(|error| {
241                    SQLError::Internal(format!("resolve sequence `{name}`: {error}"))
242                })?;
243                Ok(Some((name, relation)))
244            }
245            Value::Int(oid) => self.resolution.sequence_privilege_oid(*oid),
246            other => Err(SQLError::TypeMismatch(format!(
247                "has_sequence_privilege sequence must be text or oid, got {other:?}"
248            ))),
249        }
250    }
251
252    pub fn ensure_sequence_owner(
253        &self,
254        name: &str,
255        relation: &RelationIdentity,
256    ) -> Result<String, SQLError> {
257        let owner = self
258            .security
259            .security_read()
260            .get(relation)
261            .map(|security| security.role_owner.clone())
262            .ok_or_else(|| {
263                SQLError::Internal(format!("sequence `{name}` has no security metadata"))
264            })?;
265        crate::schema::sequences::ownership::require_sequence_ownership(&relation.name, {
266            let current = self.names.current_user_name();
267            let roles = self.roles.role_definitions();
268            let memberships = self.roles.role_memberships();
269            crate::catalog::roles::role_inherits(&roles, &memberships, &current, &owner)
270        })?;
271        Ok(owner)
272    }
273}
274
275fn resolve_sequence_privilege_role(
276    value: &Value,
277    roles: &BTreeMap<String, RoleDefinition>,
278) -> Result<Option<String>, SQLError> {
279    match value {
280        Value::Str(name) | Value::FixedChar(name) => {
281            if roles.contains_key(name) {
282                Ok(Some(name.clone()))
283            } else {
284                Err(SQLError::Routine {
285                    sqlstate: "42704".into(),
286                    message: format!("role \"{name}\" does not exist"),
287                })
288            }
289        }
290        Value::Int(oid) => Ok(roles
291            .values()
292            .find(|role| role.oid == *oid)
293            .map(|role| role.name.clone())),
294        other => Err(SQLError::TypeMismatch(format!(
295            "has_sequence_privilege role must be name or oid, got {other:?}"
296        ))),
297    }
298}
uqa_sql/catalog/security/sequence_inquiry.rs

uqa_sql/catalog/security/
sequence_inquiry.rs
