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 assignment in &statement.assignments {
46        catalog.ensure_column_privilege_for(
47            &statement.table,
48            &assignment.target.column,
49            &privilege_subject,
50            TableAclPrivilege::Update,
51        )?;
52    }
53    let expressions = statement
54        .assignments
55        .iter()
56        .flat_map(crate::plan::AssignmentPlan::expressions)
57        .chain(statement.predicate.iter())
58        .chain(
59            statement
60                .returning
61                .iter()
62                .map(|projection| &projection.expr),
63        )
64        .collect::<Vec<_>>();
65    catalog.ensure_target_select(TargetSelectPrivilegeRequest {
66        table: &statement.table,
67        privilege_subject: statement.target_privilege_subject.as_ref(),
68        target_qualifier: &statement.target_qualifier,
69        returning_aliases: &statement.returning_aliases,
70        expressions: &expressions,
71        subqueries: &statement.subqueries,
72        required_columns: &[],
73    })?;
74    Ok(expressions)
75}
76
77/// Analyze INSERT and ON CONFLICT privileges before any trigger or row mutation.
78pub fn ensure_insert_target_privileges(
79    catalog: &dyn MutationPrivilegeCatalog,
80    stmt: &crate::plan::InsertPlan,
81    conflict_update_columns: Option<&[String]>,
82) -> Result<(), SQLError> {
83    use crate::plan::{ConflictActionPlan, ConflictPlan};
84    let default_values =
85        stmt.source.is_none() && stmt.columns.is_empty() && stmt.rows.iter().all(Vec::is_empty);
86    let privilege_subject = stmt
87        .target_privilege_subject
88        .clone()
89        .unwrap_or_else(|| catalog.current_role());
90    if default_values {
91        catalog.ensure_any_column_privilege_for(
92            &stmt.table,
93            &privilege_subject,
94            TableAclPrivilege::Insert,
95        )?;
96    } else {
97        for column in insert_target_columns(catalog, stmt)? {
98            catalog.ensure_column_privilege_for(
99                &stmt.table,
100                &column,
101                &privilege_subject,
102                TableAclPrivilege::Insert,
103            )?;
104        }
105    }
106    if let Some(columns) = conflict_update_columns {
107        for column in columns {
108            catalog.ensure_column_privilege_for(
109                &stmt.table,
110                column,
111                &privilege_subject,
112                TableAclPrivilege::Update,
113            )?;
114        }
115    }
116    let mut privilege_expressions = stmt
117        .returning
118        .iter()
119        .map(|projection| &projection.expr)
120        .collect::<Vec<_>>();
121    if let Some(conflict) = &stmt.on_conflict {
122        privilege_expressions.extend(conflict.expressions.iter());
123        privilege_expressions.extend(conflict.predicate.iter().map(Box::as_ref));
124    }
125    if let Some(ConflictPlan {
126        action:
127            ConflictActionPlan::Update {
128                assignments,
129                predicate,
130            },
131        ..
132    }) = stmt.on_conflict.as_ref()
133    {
134        privilege_expressions.extend(
135            assignments
136                .iter()
137                .flat_map(crate::plan::AssignmentPlan::expressions),
138        );
139        privilege_expressions.extend(predicate.iter().map(Box::as_ref));
140    }
141    catalog.ensure_target_select(TargetSelectPrivilegeRequest {
142        table: &stmt.table,
143        privilege_subject: stmt.target_privilege_subject.as_ref(),
144        target_qualifier: &stmt.target_qualifier,
145        returning_aliases: &stmt.returning_aliases,
146        expressions: &privilege_expressions,
147        subqueries: &stmt.subqueries,
148        required_columns: stmt
149            .on_conflict
150            .as_ref()
151            .map_or(&[][..], |conflict| conflict.conflict_columns.as_slice()),
152    })?;
153    Ok(())
154}
155
156/// 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.
157pub fn insert_target_columns(
158    catalog: &dyn MutationPrivilegeCatalog,
159    stmt: &crate::plan::InsertPlan,
160) -> Result<Vec<String>, SQLError> {
161    if !stmt.columns.is_empty() {
162        return Ok(stmt
163            .columns
164            .iter()
165            .map(|target| target.column.clone())
166            .collect());
167    }
168    let supplied = stmt.source.as_deref().map_or_else(
169        || stmt.rows.first().map(Vec::len),
170        |source| {
171            crate::semantics::projection::query_plan_output_columns(source)
172                .map(|columns| columns.len())
173        },
174    );
175    if supplied == Some(0) {
176        return Ok(Vec::new());
177    }
178    let columns = catalog.bound_table_column_names(&stmt.table)?;
179    Ok(match supplied {
180        Some(supplied) => columns.into_iter().take(supplied).collect(),
181        None => columns,
182    })
183}
184
185/// Fill missing DML privilege subjects from the surrounding WITH definition.
186pub fn inherit_command_privilege_subject(
187    command: &mut crate::plan::CommandPlan,
188    subject: RoleReference,
189) -> Result<(), SQLError> {
190    use crate::plan::CommandPlan;
191    let (statement_subject, target_subject) = match command {
192        CommandPlan::Insert(plan) => (
193            &mut plan.statement_privilege_subject,
194            &mut plan.target_privilege_subject,
195        ),
196        CommandPlan::Update(plan) => (
197            &mut plan.statement_privilege_subject,
198            &mut plan.target_privilege_subject,
199        ),
200        CommandPlan::Delete(plan) => (
201            &mut plan.statement_privilege_subject,
202            &mut plan.target_privilege_subject,
203        ),
204        CommandPlan::Merge(plan) => (
205            &mut plan.statement_privilege_subject,
206            &mut plan.target_privilege_subject,
207        ),
208        _ => {
209            return Err(SQLError::Internal(
210                "non-DML command in a WITH definition".into(),
211            ))
212        }
213    };
214    statement_subject.get_or_insert(subject.clone());
215    target_subject.get_or_insert(subject);
216    Ok(())
217}
uqa_sql/semantics/mutation_privileges.rs

uqa_sql/semantics/
mutation_privileges.rs
