uqa_sql/semantics/
view_privileges.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Privilege checks at each view boundary before DML rewrite or trigger dispatch.
8
9use std::collections::BTreeSet;
10
11use crate::plan::{
12    ConflictActionPlan, ConflictPlan, DeletePlan, InsertPlan, MergePlan, MergeWhenPlan, UpdatePlan,
13};
14use crate::SQLError;
15
16use super::privileges::TargetSelectPrivilegeRequest;
17use crate::catalog::security::table::TableAclPrivilege;
18use crate::catalog::stored_view::StoredView;
19
20fn view_target(
21    services: &dyn ViewPrivilegeCatalog,
22    name: &str,
23) -> Result<(StoredView, Vec<String>), SQLError> {
24    let view = services
25        .view_definition(name)?
26        .ok_or_else(|| SQLError::UnknownTable(name.to_string()))?;
27    let columns = view.output_columns.clone().ok_or_else(|| {
28        SQLError::Internal(format!(
29            "loaded view `{name}` has no durable public column metadata"
30        ))
31    })?;
32    Ok((view, columns))
33}
34
35fn privilege_subject(
36    services: &dyn ViewPrivilegeCatalog,
37    rewritten_subject: Option<&str>,
38) -> String {
39    rewritten_subject.map_or_else(|| services.current_user_name(), str::to_string)
40}
41
42fn next_privilege_subject(view: &StoredView, subject: String) -> String {
43    if view.security_invoker() {
44        subject
45    } else {
46        view.role_owner.clone()
47    }
48}
49
50fn validate_columns(
51    name: &str,
52    available: &[String],
53    requested: &[String],
54) -> Result<(), SQLError> {
55    let mut seen = BTreeSet::new();
56    for column in requested {
57        if !seen.insert(column) {
58            return Err(SQLError::Routine {
59                sqlstate: "42701".into(),
60                message: format!("column \"{column}\" specified more than once"),
61            });
62        }
63        if !available.contains(column) {
64            return Err(SQLError::UnknownColumn(format!("{name}.{column}")));
65        }
66    }
67    Ok(())
68}
69
70pub fn ensure_insert(
71    services: &dyn ViewPrivilegeCatalog,
72    statement: &InsertPlan,
73) -> Result<String, SQLError> {
74    let (view, available) = view_target(services, &statement.table)?;
75    validate_columns(&statement.table, &available, &statement.columns)?;
76    let subject = privilege_subject(services, statement.target_privilege_subject.as_deref());
77    let default_values = statement.source.is_none()
78        && statement.columns.is_empty()
79        && statement.rows.iter().all(Vec::is_empty);
80    if default_values {
81        services.ensure_any_view_column_privilege_for(
82            &statement.table,
83            &view,
84            &subject,
85            TableAclPrivilege::Insert,
86        )?;
87    } else {
88        let columns = if statement.columns.is_empty() {
89            let supplied = statement.source.as_deref().map_or_else(
90                || statement.rows.first().map(Vec::len),
91                |source| {
92                    crate::semantics::projection::query_plan_output_columns(source)
93                        .map(|columns| columns.len())
94                },
95            );
96            supplied.map_or_else(
97                || available.clone(),
98                |width| available.iter().take(width).cloned().collect(),
99            )
100        } else {
101            statement.columns.clone()
102        };
103        for column in columns {
104            services.ensure_view_column_privilege_for(
105                &statement.table,
106                &view,
107                &column,
108                &subject,
109                TableAclPrivilege::Insert,
110            )?;
111        }
112    }
113    let mut expressions = statement
114        .returning
115        .iter()
116        .map(|projection| &projection.expr)
117        .collect::<Vec<_>>();
118    if let Some(conflict) = &statement.on_conflict {
119        expressions.extend(conflict.expressions.iter());
120        expressions.extend(conflict.predicate.iter().map(Box::as_ref));
121    }
122    let required_columns = if let Some(ConflictPlan {
123        conflict_columns,
124        action:
125            ConflictActionPlan::Update {
126                assignments,
127                predicate,
128            },
129        ..
130    }) = statement.on_conflict.as_ref()
131    {
132        let update_columns = assignments
133            .iter()
134            .map(|assignment| assignment.column.clone())
135            .collect::<Vec<_>>();
136        validate_columns(&statement.table, &available, &update_columns)?;
137        for column in &update_columns {
138            services.ensure_view_column_privilege_for(
139                &statement.table,
140                &view,
141                column,
142                &subject,
143                TableAclPrivilege::Update,
144            )?;
145        }
146        expressions.extend(assignments.iter().map(|assignment| &assignment.value));
147        expressions.extend(predicate.iter().map(Box::as_ref));
148        conflict_columns.as_slice()
149    } else {
150        &[]
151    };
152    services.ensure_target_select(TargetSelectPrivilegeRequest {
153        table: &statement.table,
154        privilege_subject: Some(&subject),
155        target_qualifier: &statement.target_qualifier,
156        returning_aliases: &statement.returning_aliases,
157        expressions: &expressions,
158        subqueries: &statement.subqueries,
159        required_columns,
160    })?;
161    Ok(next_privilege_subject(&view, subject))
162}
163
164pub fn ensure_update(
165    services: &dyn ViewPrivilegeCatalog,
166    statement: &UpdatePlan,
167) -> Result<String, SQLError> {
168    let (view, available) = view_target(services, &statement.table)?;
169    let columns = statement
170        .assignments
171        .iter()
172        .map(|assignment| assignment.column.clone())
173        .collect::<Vec<_>>();
174    validate_columns(&statement.table, &available, &columns)?;
175    let subject = privilege_subject(services, statement.target_privilege_subject.as_deref());
176    for column in &columns {
177        services.ensure_view_column_privilege_for(
178            &statement.table,
179            &view,
180            column,
181            &subject,
182            TableAclPrivilege::Update,
183        )?;
184    }
185    let expressions = statement
186        .assignments
187        .iter()
188        .map(|assignment| &assignment.value)
189        .chain(statement.predicate.iter())
190        .chain(
191            statement
192                .returning
193                .iter()
194                .map(|projection| &projection.expr),
195        )
196        .collect::<Vec<_>>();
197    services.ensure_target_select(TargetSelectPrivilegeRequest {
198        table: &statement.table,
199        privilege_subject: Some(&subject),
200        target_qualifier: &statement.target_qualifier,
201        returning_aliases: &statement.returning_aliases,
202        expressions: &expressions,
203        subqueries: &statement.subqueries,
204        required_columns: &[],
205    })?;
206    Ok(next_privilege_subject(&view, subject))
207}
208
209pub fn ensure_delete(
210    services: &dyn ViewPrivilegeCatalog,
211    statement: &DeletePlan,
212) -> Result<String, SQLError> {
213    let (view, _) = view_target(services, &statement.table)?;
214    let subject = privilege_subject(services, statement.target_privilege_subject.as_deref());
215    services.ensure_view_privilege_for(
216        &statement.table,
217        &view,
218        &subject,
219        TableAclPrivilege::Delete,
220    )?;
221    let expressions = statement
222        .predicate
223        .iter()
224        .chain(
225            statement
226                .returning
227                .iter()
228                .map(|projection| &projection.expr),
229        )
230        .collect::<Vec<_>>();
231    services.ensure_target_select(TargetSelectPrivilegeRequest {
232        table: &statement.table,
233        privilege_subject: Some(&subject),
234        target_qualifier: &statement.target_qualifier,
235        returning_aliases: &statement.returning_aliases,
236        expressions: &expressions,
237        subqueries: &statement.subqueries,
238        required_columns: &[],
239    })?;
240    Ok(next_privilege_subject(&view, subject))
241}
242
243pub fn ensure_merge(
244    services: &dyn ViewPrivilegeCatalog,
245    statement: &MergePlan,
246) -> Result<String, SQLError> {
247    let (view, available) = view_target(services, &statement.target)?;
248    let subject = privilege_subject(services, statement.target_privilege_subject.as_deref());
249    let mut requires_delete = false;
250    let mut requires_any_insert = false;
251    let mut column_privileges = BTreeSet::new();
252    for clause in &statement.when_clauses {
253        match clause {
254            MergeWhenPlan::InsertNotMatched {
255                columns, values, ..
256            } => {
257                validate_columns(&statement.target, &available, columns)?;
258                if columns.is_empty() && values.is_empty() {
259                    requires_any_insert = true;
260                } else {
261                    let columns = if columns.is_empty() {
262                        available.iter().take(values.len()).cloned().collect()
263                    } else {
264                        columns.clone()
265                    };
266                    column_privileges.extend(
267                        columns
268                            .into_iter()
269                            .map(|column| (TableAclPrivilege::Insert, column)),
270                    );
271                }
272            }
273            MergeWhenPlan::UpdateMatched { assignments, .. }
274            | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
275                let columns = assignments
276                    .iter()
277                    .map(|assignment| assignment.column.clone())
278                    .collect::<Vec<_>>();
279                validate_columns(&statement.target, &available, &columns)?;
280                column_privileges.extend(
281                    columns
282                        .into_iter()
283                        .map(|column| (TableAclPrivilege::Update, column)),
284                );
285            }
286            MergeWhenPlan::DeleteMatched { .. }
287            | MergeWhenPlan::DeleteNotMatchedBySource { .. } => requires_delete = true,
288            _ => {}
289        }
290    }
291    if requires_delete {
292        services.ensure_view_privilege_for(
293            &statement.target,
294            &view,
295            &subject,
296            TableAclPrivilege::Delete,
297        )?;
298    }
299    if requires_any_insert {
300        services.ensure_any_view_column_privilege_for(
301            &statement.target,
302            &view,
303            &subject,
304            TableAclPrivilege::Insert,
305        )?;
306    }
307    for (privilege, column) in column_privileges {
308        services.ensure_view_column_privilege_for(
309            &statement.target,
310            &view,
311            &column,
312            &subject,
313            privilege,
314        )?;
315    }
316    let expressions = merge_privilege_expressions(statement);
317    services.ensure_target_select(TargetSelectPrivilegeRequest {
318        table: &statement.target,
319        privilege_subject: Some(&subject),
320        target_qualifier: &statement.target_qualifier,
321        returning_aliases: &statement.returning_aliases,
322        expressions: &expressions,
323        subqueries: &statement.subqueries,
324        required_columns: &[],
325    })?;
326    Ok(next_privilege_subject(&view, subject))
327}
328
329pub fn merge_privilege_expressions(stmt: &MergePlan) -> Vec<&crate::ScalarExpr> {
330    let mut expressions = vec![&stmt.join_condition];
331    expressions.extend(stmt.target_predicate.iter());
332    expressions.extend(stmt.returning.iter().map(|projection| &projection.expr));
333    for clause in &stmt.when_clauses {
334        match clause {
335            MergeWhenPlan::UpdateMatched {
336                condition,
337                assignments,
338            }
339            | MergeWhenPlan::UpdateNotMatchedBySource {
340                condition,
341                assignments,
342            } => {
343                expressions.extend(condition.iter());
344                expressions.extend(assignments.iter().map(|assignment| &assignment.value));
345            }
346            MergeWhenPlan::InsertNotMatched {
347                condition, values, ..
348            } => {
349                expressions.extend(condition.iter());
350                expressions.extend(values);
351            }
352            MergeWhenPlan::DeleteMatched { condition }
353            | MergeWhenPlan::DeleteNotMatchedBySource { condition }
354            | MergeWhenPlan::NothingMatched { condition }
355            | MergeWhenPlan::NothingNotMatched { condition }
356            | MergeWhenPlan::NothingNotMatchedBySource { condition } => {
357                expressions.extend(condition.iter());
358            }
359        }
360    }
361    expressions
362}
363
364/// Authorization access to one loaded view and the current statement's SELECT privileges.
365pub trait ViewPrivilegeCatalog {
366    fn view_definition(&self, name: &str) -> Result<Option<StoredView>, SQLError>;
367    fn current_user_name(&self) -> String;
368    fn ensure_view_privilege_for(
369        &self,
370        name: &str,
371        view: &StoredView,
372        subject: &str,
373        privilege: TableAclPrivilege,
374    ) -> Result<(), SQLError>;
375    fn ensure_view_column_privilege_for(
376        &self,
377        name: &str,
378        view: &StoredView,
379        column: &str,
380        subject: &str,
381        privilege: TableAclPrivilege,
382    ) -> Result<(), SQLError>;
383    fn ensure_any_view_column_privilege_for(
384        &self,
385        name: &str,
386        view: &StoredView,
387        subject: &str,
388        privilege: TableAclPrivilege,
389    ) -> Result<(), SQLError>;
390    fn ensure_target_select(
391        &self,
392        request: TargetSelectPrivilegeRequest<'_, '_>,
393    ) -> Result<(), SQLError>;
394}
uqa_sql/semantics/view_privileges.rs

uqa_sql/semantics/
view_privileges.rs
