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

uqa_sql/catalog/security/
table_inquiry.rs
