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

uqa_sql/semantics/
view_privileges.rs
