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        let insert_columns = if stmt.columns.is_empty() {
98            let supplied = stmt.source.as_deref().map_or_else(
99                || stmt.rows.first().map(Vec::len),
100                |source| {
101                    crate::semantics::projection::query_plan_output_columns(source)
102                        .map(|columns| columns.len())
103                },
104            );
105            let columns = catalog.bound_table_column_names(&stmt.table)?;
106            match supplied {
107                Some(supplied) => columns.into_iter().take(supplied).collect(),
108                None => columns,
109            }
110        } else {
111            stmt.columns
112                .iter()
113                .map(|target| target.column.clone())
114                .collect()
115        };
116        for column in insert_columns {
117            catalog.ensure_column_privilege_for(
118                &stmt.table,
119                &column,
120                &privilege_subject,
121                TableAclPrivilege::Insert,
122            )?;
123        }
124    }
125    if let Some(columns) = conflict_update_columns {
126        for column in columns {
127            catalog.ensure_column_privilege_for(
128                &stmt.table,
129                column,
130                &privilege_subject,
131                TableAclPrivilege::Update,
132            )?;
133        }
134    }
135    let mut privilege_expressions = stmt
136        .returning
137        .iter()
138        .map(|projection| &projection.expr)
139        .collect::<Vec<_>>();
140    if let Some(conflict) = &stmt.on_conflict {
141        privilege_expressions.extend(conflict.expressions.iter());
142        privilege_expressions.extend(conflict.predicate.iter().map(Box::as_ref));
143    }
144    if let Some(ConflictPlan {
145        action:
146            ConflictActionPlan::Update {
147                assignments,
148                predicate,
149            },
150        ..
151    }) = stmt.on_conflict.as_ref()
152    {
153        privilege_expressions.extend(
154            assignments
155                .iter()
156                .flat_map(crate::plan::AssignmentPlan::expressions),
157        );
158        privilege_expressions.extend(predicate.iter().map(Box::as_ref));
159    }
160    catalog.ensure_target_select(TargetSelectPrivilegeRequest {
161        table: &stmt.table,
162        privilege_subject: stmt.target_privilege_subject.as_ref(),
163        target_qualifier: &stmt.target_qualifier,
164        returning_aliases: &stmt.returning_aliases,
165        expressions: &privilege_expressions,
166        subqueries: &stmt.subqueries,
167        required_columns: stmt
168            .on_conflict
169            .as_ref()
170            .map_or(&[][..], |conflict| conflict.conflict_columns.as_slice()),
171    })?;
172    Ok(())
173}
174
175/// Fill missing DML privilege subjects from the surrounding WITH definition.
176pub fn inherit_command_privilege_subject(
177    command: &mut crate::plan::CommandPlan,
178    subject: RoleReference,
179) -> Result<(), SQLError> {
180    use crate::plan::CommandPlan;
181    let (statement_subject, target_subject) = match command {
182        CommandPlan::Insert(plan) => (
183            &mut plan.statement_privilege_subject,
184            &mut plan.target_privilege_subject,
185        ),
186        CommandPlan::Update(plan) => (
187            &mut plan.statement_privilege_subject,
188            &mut plan.target_privilege_subject,
189        ),
190        CommandPlan::Delete(plan) => (
191            &mut plan.statement_privilege_subject,
192            &mut plan.target_privilege_subject,
193        ),
194        CommandPlan::Merge(plan) => (
195            &mut plan.statement_privilege_subject,
196            &mut plan.target_privilege_subject,
197        ),
198        _ => {
199            return Err(SQLError::Internal(
200                "non-DML command in a WITH definition".into(),
201            ))
202        }
203    };
204    statement_subject.get_or_insert(subject.clone());
205    target_subject.get_or_insert(subject);
206    Ok(())
207}
uqa_sql/semantics/mutation_privileges.rs

uqa_sql/semantics/
mutation_privileges.rs
