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

uqa_sql/semantics/
view_privileges.rs
