uqa_sql/semantics/
mutation_privileges.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Mutation target privilege analysis.
8use crate::catalog::roles::RoleReference;
9use crate::{
10    catalog::security::table::TableAclPrivilege,
11    plan::UpdatePlan,
12    semantics::{privileges::TargetSelectPrivilegeRequest, view_privileges::ViewPrivilegeCatalog},
13    SQLError, ScalarExpr,
14};
15pub trait MutationPrivilegeCatalog: ViewPrivilegeCatalog {
16    fn bound_table_column_names(&self, table: &str) -> Result<Vec<String>, SQLError>;
17    fn ensure_table_privilege_for(
18        &self,
19        table: &str,
20        subject: &RoleReference,
21        privilege: TableAclPrivilege,
22    ) -> Result<(), SQLError>;
23    fn ensure_column_privilege_for(
24        &self,
25        table: &str,
26        column: &str,
27        subject: &RoleReference,
28        privilege: TableAclPrivilege,
29    ) -> Result<(), SQLError>;
30    fn ensure_any_column_privilege_for(
31        &self,
32        table: &str,
33        subject: &RoleReference,
34        privilege: TableAclPrivilege,
35    ) -> Result<(), SQLError>;
36}
37pub fn ensure_update_target_privileges<'a>(
38    catalog: &dyn MutationPrivilegeCatalog,
39    statement: &'a UpdatePlan,
40) -> Result<Vec<&'a ScalarExpr>, SQLError> {
41    let privilege_subject = statement
42        .target_privilege_subject
43        .clone()
44        .unwrap_or_else(|| catalog.current_role());
45    for target in statement
46        .assignments
47        .iter()
48        .flat_map(|assignment| assignment.target.targets())
49    {
50        catalog.ensure_column_privilege_for(
51            &statement.table,
52            &target.column,
53            &privilege_subject,
54            TableAclPrivilege::Update,
55        )?;
56    }
57    let expressions = statement
58        .assignments
59        .iter()
60        .flat_map(crate::plan::AssignmentPlan::expressions)
61        .chain(statement.predicate.iter())
62        .chain(
63            statement
64                .returning
65                .iter()
66                .map(|projection| &projection.expr),
67        )
68        .collect::<Vec<_>>();
69    catalog.ensure_target_select(TargetSelectPrivilegeRequest {
70        table: &statement.table,
71        privilege_subject: statement.target_privilege_subject.as_ref(),
72        target_qualifier: &statement.target_qualifier,
73        returning_aliases: &statement.returning_aliases,
74        expressions: &expressions,
75        subqueries: &statement.subqueries,
76        required_columns: &[],
77    })?;
78    Ok(expressions)
79}
80
81/// Analyze INSERT and ON CONFLICT privileges before any trigger or row mutation.
82pub fn ensure_insert_target_privileges(
83    catalog: &dyn MutationPrivilegeCatalog,
84    stmt: &crate::plan::InsertPlan,
85    conflict_update_columns: Option<&[String]>,
86) -> Result<(), SQLError> {
87    use crate::plan::{ConflictActionPlan, ConflictPlan};
88    let default_values =
89        stmt.source.is_none() && stmt.columns.is_empty() && stmt.rows.iter().all(Vec::is_empty);
90    let privilege_subject = stmt
91        .target_privilege_subject
92        .clone()
93        .unwrap_or_else(|| catalog.current_role());
94    if default_values {
95        catalog.ensure_any_column_privilege_for(
96            &stmt.table,
97            &privilege_subject,
98            TableAclPrivilege::Insert,
99        )?;
100    } else {
101        for column in insert_target_columns(catalog, stmt)? {
102            catalog.ensure_column_privilege_for(
103                &stmt.table,
104                &column,
105                &privilege_subject,
106                TableAclPrivilege::Insert,
107            )?;
108        }
109    }
110    if let Some(columns) = conflict_update_columns {
111        for column in columns {
112            catalog.ensure_column_privilege_for(
113                &stmt.table,
114                column,
115                &privilege_subject,
116                TableAclPrivilege::Update,
117            )?;
118        }
119    }
120    let mut privilege_expressions = stmt
121        .returning
122        .iter()
123        .map(|projection| &projection.expr)
124        .collect::<Vec<_>>();
125    if let Some(conflict) = &stmt.on_conflict {
126        privilege_expressions.extend(conflict.expressions.iter());
127        privilege_expressions.extend(conflict.predicate.iter().map(Box::as_ref));
128    }
129    if let Some(ConflictPlan {
130        action:
131            ConflictActionPlan::Update {
132                assignments,
133                predicate,
134            },
135        ..
136    }) = stmt.on_conflict.as_ref()
137    {
138        privilege_expressions.extend(
139            assignments
140                .iter()
141                .flat_map(crate::plan::AssignmentPlan::expressions),
142        );
143        privilege_expressions.extend(predicate.iter().map(Box::as_ref));
144    }
145    catalog.ensure_target_select(TargetSelectPrivilegeRequest {
146        table: &stmt.table,
147        privilege_subject: stmt.target_privilege_subject.as_ref(),
148        target_qualifier: &stmt.target_qualifier,
149        returning_aliases: &stmt.returning_aliases,
150        expressions: &privilege_expressions,
151        subqueries: &stmt.subqueries,
152        required_columns: stmt
153            .on_conflict
154            .as_ref()
155            .map_or(&[][..], |conflict| conflict.conflict_columns.as_slice()),
156    })?;
157    Ok(())
158}
159
160/// The columns an `INSERT` supplies, as `PostgreSQL` records them in its target's `insertedCols`: the columns it names, or else the leading columns of the table, one for each value a row supplies; `DEFAULT VALUES` supplies none.
161pub fn insert_target_columns(
162    catalog: &dyn MutationPrivilegeCatalog,
163    stmt: &crate::plan::InsertPlan,
164) -> Result<Vec<String>, SQLError> {
165    insert_supplied_columns(stmt, || catalog.bound_table_column_names(&stmt.table))
166}
167
168/// The columns an `INSERT` supplies, as [`insert_target_columns`] finds them, for a target whose columns `target_columns` lists in order.
169pub fn insert_supplied_columns(
170    stmt: &crate::plan::InsertPlan,
171    target_columns: impl FnOnce() -> Result<Vec<String>, SQLError>,
172) -> Result<Vec<String>, SQLError> {
173    if !stmt.columns.is_empty() {
174        return Ok(stmt
175            .columns
176            .iter()
177            .map(|target| target.column.clone())
178            .collect());
179    }
180    let supplied = stmt.source.as_deref().map_or_else(
181        || stmt.rows.first().map(Vec::len),
182        |source| {
183            crate::semantics::projection::query_plan_output_columns(source)
184                .map(|columns| columns.len())
185        },
186    );
187    if supplied == Some(0) {
188        return Ok(Vec::new());
189    }
190    let columns = target_columns()?;
191    Ok(match supplied {
192        Some(supplied) => columns.into_iter().take(supplied).collect(),
193        None => columns,
194    })
195}
196
197/// Fill missing DML privilege subjects from the surrounding WITH definition.
198pub fn inherit_command_privilege_subject(
199    command: &mut crate::plan::CommandPlan,
200    subject: RoleReference,
201) -> Result<(), SQLError> {
202    use crate::plan::CommandPlan;
203    let (statement_subject, target_subject) = match command {
204        CommandPlan::Insert(plan) => (
205            &mut plan.statement_privilege_subject,
206            &mut plan.target_privilege_subject,
207        ),
208        CommandPlan::Update(plan) => (
209            &mut plan.statement_privilege_subject,
210            &mut plan.target_privilege_subject,
211        ),
212        CommandPlan::Delete(plan) => (
213            &mut plan.statement_privilege_subject,
214            &mut plan.target_privilege_subject,
215        ),
216        CommandPlan::Merge(plan) => (
217            &mut plan.statement_privilege_subject,
218            &mut plan.target_privilege_subject,
219        ),
220        _ => {
221            return Err(SQLError::Internal(
222                "non-DML command in a WITH definition".into(),
223            ))
224        }
225    };
226    statement_subject.get_or_insert(subject.clone());
227    target_subject.get_or_insert(subject);
228    Ok(())
229}
230
231/// Authorize a DELETE target and the columns read by its predicate and RETURNING.
232pub fn ensure_delete_target_privileges<'a>(
233    catalog: &dyn MutationPrivilegeCatalog,
234    statement: &'a crate::plan::DeletePlan,
235) -> Result<Vec<&'a ScalarExpr>, SQLError> {
236    let privilege_subject = statement
237        .target_privilege_subject
238        .clone()
239        .unwrap_or_else(|| catalog.current_role());
240    catalog.ensure_table_privilege_for(
241        &statement.table,
242        &privilege_subject,
243        TableAclPrivilege::Delete,
244    )?;
245    let privilege_expressions = statement
246        .predicate
247        .iter()
248        .chain(
249            statement
250                .returning
251                .iter()
252                .map(|projection| &projection.expr),
253        )
254        .collect::<Vec<_>>();
255    catalog.ensure_target_select(TargetSelectPrivilegeRequest {
256        table: &statement.table,
257        privilege_subject: statement.target_privilege_subject.as_ref(),
258        target_qualifier: &statement.target_qualifier,
259        returning_aliases: &statement.returning_aliases,
260        expressions: &privilege_expressions,
261        subqueries: &statement.subqueries,
262        required_columns: &[],
263    })?;
264    Ok(privilege_expressions)
265}
uqa_sql/semantics/mutation_privileges.rs

uqa_sql/semantics/
mutation_privileges.rs
