uqa_sql/catalog/security/
table_inquiry.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Table and column privilege inquiry and `PostgreSQL` relation/attribute binding rules.
8
9use super::{
10    columns::role_has_column_privilege as column_privilege_check,
11    sequence_inquiry::SequenceTablePrivilegeInquiry,
12    table::{
13        parse_column_privilege_checks, parse_privilege_checks, role_has_privilege,
14        TablePrivilegeCheck,
15    },
16    TableSecurity,
17};
18use crate::catalog::roles::{identity::RoleSubject, RoleReference};
19use crate::{
20    catalog::{
21        resolution::RelationResolution,
22        roles::{guards::RoleCatalogGuards, RoleDefinition, RoleReferenceNames},
23    },
24    SQLError,
25};
26use std::collections::BTreeMap;
27use uqa_core::{catalog_acl::AclGrantee, RelationIdentity, Value};
28
29pub trait TablePrivilegeCatalog {
30    fn visible_relation_kind(&self, reference: &str) -> Result<RelationResolution, SQLError>;
31    fn resolve_table_privilege_oid(
32        &self,
33        oid: i64,
34    ) -> Result<Option<ResolvedTablePrivilegeTarget>, SQLError>;
35    fn table_privilege_security(
36        &self,
37        target: &ResolvedTablePrivilegeTarget,
38        roles: &BTreeMap<String, RoleDefinition>,
39    ) -> Result<TableSecurity, SQLError>;
40    fn column_privilege_relation(
41        &self,
42        target: &ResolvedTablePrivilegeTarget,
43        roles: &BTreeMap<String, RoleDefinition>,
44    ) -> Result<ColumnPrivilegeRelation, SQLError>;
45}
46
47pub struct TablePrivilegeInquiry<'a> {
48    pub names: &'a dyn RoleReferenceNames,
49    pub roles: &'a dyn RoleCatalogGuards,
50    pub sequences: &'a dyn SequenceTablePrivilegeInquiry,
51    pub catalog: &'a dyn TablePrivilegeCatalog,
52}
53
54pub enum ResolvedTablePrivilegeTarget {
55    System(crate::catalog::SystemRelation),
56    Table(RelationIdentity),
57    View(RelationIdentity),
58    ForeignTable(RelationIdentity),
59    Sequence(RelationIdentity),
60}
61
62enum ResolvedColumnPrivilegeTarget {
63    User(String),
64    System,
65}
66
67const POSTGRES_SYSTEM_COLUMNS: [&str; 6] = ["ctid", "xmin", "cmin", "xmax", "cmax", "tableoid"];
68
69pub struct ColumnPrivilegeRelation {
70    pub relation: RelationIdentity,
71    pub security: TableSecurity,
72    pub columns: Vec<String>,
73    pub has_system_columns: bool,
74}
75
76fn resolve_column_privilege_target(
77    relation: &RelationIdentity,
78    columns: &[String],
79    has_system_columns: bool,
80    value: &Value,
81) -> Result<Option<ResolvedColumnPrivilegeTarget>, SQLError> {
82    match value {
83        Value::Str(column) | Value::FixedChar(column) => {
84            if columns.iter().any(|definition| definition == column) {
85                Ok(Some(ResolvedColumnPrivilegeTarget::User(column.clone())))
86            } else if has_system_columns && POSTGRES_SYSTEM_COLUMNS.contains(&column.as_str()) {
87                Ok(Some(ResolvedColumnPrivilegeTarget::System))
88            } else {
89                Err(SQLError::Routine {
90                    sqlstate: "42703".into(),
91                    message: format!(
92                        "column \"{column}\" of relation \"{}\" does not exist",
93                        relation.name
94                    ),
95                })
96            }
97        }
98        Value::Int(attnum) if *attnum > 0 => Ok(usize::try_from(*attnum - 1)
99            .ok()
100            .and_then(|index| columns.get(index))
101            .map(|column| ResolvedColumnPrivilegeTarget::User(column.clone()))),
102        Value::Int(attnum) if has_system_columns && (-6..=-1).contains(attnum) => {
103            Ok(Some(ResolvedColumnPrivilegeTarget::System))
104        }
105        Value::Int(_) => Ok(None),
106        other => Err(SQLError::TypeMismatch(format!(
107            "has_column_privilege column must be text or smallint, got {other:?}"
108        ))),
109    }
110}
111
112fn privilege_text<'a>(function: &str, value: &'a Value) -> Result<&'a str, SQLError> {
113    match value {
114        Value::Str(privilege) | Value::FixedChar(privilege) => Ok(privilege),
115        other => Err(SQLError::TypeMismatch(format!(
116            "{function} privilege must be text, got {other:?}"
117        ))),
118    }
119}
120
121fn table_privilege_checks(value: &Value) -> Result<Vec<TablePrivilegeCheck>, SQLError> {
122    parse_privilege_checks(privilege_text("has_table_privilege", value)?)
123}
124
125fn column_privilege_checks(value: &Value) -> Result<Vec<TablePrivilegeCheck>, SQLError> {
126    parse_column_privilege_checks(privilege_text("has_column_privilege", value)?)
127}
128
129fn column_privilege_arguments(
130    arguments: &[Value],
131) -> Result<(Option<&Value>, &Value, &Value, &Value), SQLError> {
132    match arguments {
133        [table, column, privilege] => Ok((None, table, column, privilege)),
134        [subject, table, column, privilege] => Ok((Some(subject), table, column, privilege)),
135        _ => Err(SQLError::BadArity {
136            name: "has_column_privilege".into(),
137            expected: "3 or 4".into(),
138            actual: arguments.len(),
139        }),
140    }
141}
142
143impl TablePrivilegeInquiry<'_> {
144    pub fn has_table_privilege_value(&self, arguments: &[Value]) -> Result<Value, SQLError> {
145        if arguments.iter().any(|argument| argument == &Value::Null) {
146            return Ok(Value::Null);
147        }
148        let (subject_value, table_value, privilege_value) = match arguments {
149            [table, privilege] => (None, table, privilege),
150            [subject, table, privilege] => (Some(subject), table, privilege),
151            _ => {
152                return Err(SQLError::BadArity {
153                    name: "has_table_privilege".into(),
154                    expected: "2 or 3".into(),
155                    actual: arguments.len(),
156                })
157            }
158        };
159        let subject = self.bind_subject(subject_value, "has_table_privilege")?;
160        let subject: &dyn RoleSubject = subject
161            .as_ref()
162            .map_or(&AclGrantee::Public as &dyn RoleSubject, |subject| {
163                subject as &dyn RoleSubject
164            });
165        let checks = matches!(table_value, Value::Int(_))
166            .then(|| table_privilege_checks(privilege_value))
167            .transpose()?;
168        let Some(target) = self.resolve_table_privilege_target(table_value)? else {
169            return Ok(Value::Null);
170        };
171        let checks = checks.map_or_else(|| table_privilege_checks(privilege_value), Ok)?;
172        if let ResolvedTablePrivilegeTarget::Sequence(relation) = &target {
173            return self
174                .sequences
175                .sequence_table_privileges(relation, subject, &checks)
176                .map(Value::Bool);
177        }
178        let roles = self.roles.role_definitions();
179        let security = self.catalog.table_privilege_security(&target, &roles)?;
180        let memberships = self.roles.role_memberships();
181        Ok(Value::Bool(checks.into_iter().any(|check| {
182            if let ResolvedTablePrivilegeTarget::System(relation) = target {
183                return super::system_relations::has_table_privilege(
184                    relation,
185                    &security,
186                    subject,
187                    check,
188                    &roles,
189                    &memberships,
190                );
191            }
192            role_has_privilege(&security, subject, check, &roles, &memberships)
193        })))
194    }
195
196    pub fn has_column_privilege_value(&self, arguments: &[Value]) -> Result<Value, SQLError> {
197        if arguments.iter().any(|argument| argument == &Value::Null) {
198            return Ok(Value::Null);
199        }
200        let (subject_value, table_value, column_value, privilege_value) =
201            column_privilege_arguments(arguments)?;
202        let subject = self.bind_subject(subject_value, "has_column_privilege")?;
203        let subject: &dyn RoleSubject = subject
204            .as_ref()
205            .map_or(&AclGrantee::Public as &dyn RoleSubject, |subject| {
206                subject as &dyn RoleSubject
207            });
208        let mut checks = (matches!(table_value, Value::Int(_))
209            && matches!(column_value, Value::Int(_)))
210        .then(|| column_privilege_checks(privilege_value))
211        .transpose()?;
212        let Some(target) = self.resolve_table_privilege_target(table_value)? else {
213            if checks.is_none() {
214                column_privilege_checks(privilege_value)?;
215            }
216            return Ok(Value::Null);
217        };
218        if checks.is_none() && matches!(column_value, Value::Int(_)) {
219            checks = Some(column_privilege_checks(privilege_value)?);
220        }
221        if let ResolvedTablePrivilegeTarget::Sequence(relation) = &target {
222            return self.has_sequence_column_privilege_value(
223                relation,
224                subject,
225                column_value,
226                privilege_value,
227                checks,
228            );
229        }
230        let roles = self.roles.role_definitions();
231        let metadata = self.catalog.column_privilege_relation(&target, &roles)?;
232        let column = resolve_column_privilege_target(
233            &metadata.relation,
234            &metadata.columns,
235            metadata.has_system_columns,
236            column_value,
237        )?;
238        let checks = checks.map_or_else(|| column_privilege_checks(privilege_value), Ok)?;
239        let Some(column) = column else {
240            return Ok(Value::Null);
241        };
242        let memberships = self.roles.role_memberships();
243        Ok(Value::Bool(checks.into_iter().any(|check| {
244            if let ResolvedTablePrivilegeTarget::System(relation) = target {
245                return match &column {
246                    ResolvedColumnPrivilegeTarget::User(column) => {
247                        super::system_relations::has_column_privilege(
248                            relation,
249                            &metadata.security,
250                            column,
251                            subject,
252                            check,
253                            &roles,
254                            &memberships,
255                        )
256                    }
257                    ResolvedColumnPrivilegeTarget::System => {
258                        super::system_relations::has_table_privilege(
259                            relation,
260                            &metadata.security,
261                            subject,
262                            check,
263                            &roles,
264                            &memberships,
265                        )
266                    }
267                };
268            }
269            match &column {
270                ResolvedColumnPrivilegeTarget::User(column) => column_privilege_check(
271                    &metadata.security,
272                    column,
273                    subject,
274                    check,
275                    &roles,
276                    &memberships,
277                ),
278                ResolvedColumnPrivilegeTarget::System => {
279                    role_has_privilege(&metadata.security, subject, check, &roles, &memberships)
280                }
281            }
282        })))
283    }
284
285    fn has_sequence_column_privilege_value(
286        &self,
287        relation: &RelationIdentity,
288        subject: &dyn RoleSubject,
289        column_value: &Value,
290        privilege_value: &Value,
291        checks: Option<Vec<TablePrivilegeCheck>>,
292    ) -> Result<Value, SQLError> {
293        let valid_column = match column_value {
294            Value::Str(column) | Value::FixedChar(column) => {
295                matches!(column.as_str(), "last_value" | "log_cnt" | "is_called")
296                    || POSTGRES_SYSTEM_COLUMNS.contains(&column.as_str())
297            }
298            Value::Int(attnum) => (1..=3).contains(attnum) || (-6..=-1).contains(attnum),
299            other => {
300                return Err(SQLError::TypeMismatch(format!(
301                    "has_column_privilege column must be text or smallint, got {other:?}"
302                )))
303            }
304        };
305        if !valid_column {
306            if let Value::Str(column) | Value::FixedChar(column) = column_value {
307                return Err(SQLError::Routine {
308                    sqlstate: "42703".into(),
309                    message: format!(
310                        "column \"{column}\" of relation \"{}\" does not exist",
311                        relation.name
312                    ),
313                });
314            }
315        }
316        let checks = checks.map_or_else(|| column_privilege_checks(privilege_value), Ok)?;
317        if !valid_column {
318            return Ok(Value::Null);
319        }
320        self.sequences
321            .sequence_table_privileges(relation, subject, &checks)
322            .map(Value::Bool)
323    }
324
325    fn bind_subject(
326        &self,
327        value: Option<&Value>,
328        function: &str,
329    ) -> Result<Option<RoleReference>, SQLError> {
330        match value {
331            None => Ok(Some(self.names.current_role())),
332            Some(value) => {
333                let roles = self.roles.inquiry_role_definitions()?;
334                super::role_bindings::bind_inquiry_subject(value, &roles, function)
335            }
336        }
337    }
338
339    fn resolve_table_privilege_target(
340        &self,
341        value: &Value,
342    ) -> Result<Option<ResolvedTablePrivilegeTarget>, SQLError> {
343        match value {
344            Value::Str(reference) | Value::FixedChar(reference) => {
345                let (name, kind) = match self.catalog.visible_relation_kind(reference)? {
346                    RelationResolution::Found(name, kind) => (name, kind),
347                    RelationResolution::MissingSchema(schema) => {
348                        return Err(SQLError::Routine {
349                            sqlstate: "3F000".into(),
350                            message: format!("schema \"{schema}\" does not exist"),
351                        })
352                    }
353                    RelationResolution::MissingRelation => {
354                        return Err(SQLError::Routine {
355                            sqlstate: "42P01".into(),
356                            message: format!("relation \"{reference}\" does not exist"),
357                        })
358                    }
359                };
360                if let Some(relation) = crate::catalog::SystemRelation::from_qualified_name(&name) {
361                    return Ok(Some(ResolvedTablePrivilegeTarget::System(relation)));
362                }
363                if !matches!(
364                    kind,
365                    "table" | "view" | "materialized view" | "foreign table" | "sequence"
366                ) {
367                    return Err(SQLError::Unsupported(format!(
368                        "has_table_privilege for {kind} is not supported"
369                    )));
370                }
371                let relation = RelationIdentity::from_legacy_name(&name).map_err(|error| {
372                    SQLError::Internal(format!("resolve table `{name}`: {error}"))
373                })?;
374                Ok(Some(match kind {
375                    "table" => ResolvedTablePrivilegeTarget::Table(relation),
376                    "view" | "materialized view" => ResolvedTablePrivilegeTarget::View(relation),
377                    "foreign table" => ResolvedTablePrivilegeTarget::ForeignTable(relation),
378                    "sequence" => ResolvedTablePrivilegeTarget::Sequence(relation),
379                    _ => unreachable!("relation kind was validated above"),
380                }))
381            }
382            Value::Int(oid) => self.catalog.resolve_table_privilege_oid(*oid),
383            other => Err(SQLError::TypeMismatch(format!(
384                "has_table_privilege table must be text or oid, got {other:?}"
385            ))),
386        }
387    }
388}
389
390#[cfg(test)]
391mod tests;
uqa_sql/catalog/security/table_inquiry.rs

uqa_sql/catalog/security/
table_inquiry.rs
