uqa-sql 0.3.5

PostgreSQL-compatible SQL compiler built on libpg_query
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

use super::{context::PrivilegeScope, BaseColumn, SourceLineage};
use crate::{catalog::resolution::RelationLookupMode, SQLError};
use std::collections::BTreeSet;

pub(super) fn ensure_required_select(
    lineage: &SourceLineage,
    required: &BTreeSet<BaseColumn>,
    scope: &PrivilegeScope<'_>,
) -> Result<(), SQLError> {
    let mut resolution = scope.resolution.clone();
    resolution.set_lookup_mode(RelationLookupMode::Bound);
    let subject = scope.privilege_subject()?;
    for (table, columns) in &lineage.tables {
        let relation = scope
            .catalog
            .relation(&resolution, table)?
            .ok_or_else(|| SQLError::UnknownTable(table.clone()))?;
        let table_required = required
            .iter()
            .filter(|column| &column.table == table)
            .map(|column| column.column.as_str())
            .collect::<BTreeSet<_>>();
        if table_required.is_empty() {
            let mut permitted =
                scope
                    .catalog
                    .has_select_privilege(&resolution, &relation, None, subject)?;
            // Foreign relation ACL lookup can report metadata errors even after a table grant.
            let inspect_columns = !permitted
                || matches!(
                    relation.kind,
                    super::context::PrivilegeRelationKind::ForeignTable
                );
            if inspect_columns {
                for column in columns {
                    if scope.catalog.has_select_privilege(
                        &resolution,
                        &relation,
                        Some(column),
                        subject,
                    )? {
                        permitted = true;
                        break;
                    }
                }
            }
            if !permitted {
                return relation_permission_denied(table, relation.kind.description());
            }
        } else {
            for column in table_required {
                let column = columns.contains(column).then_some(column);
                if !scope
                    .catalog
                    .has_select_privilege(&resolution, &relation, column, subject)?
                {
                    return relation_permission_denied(table, relation.kind.description());
                }
            }
        }
    }
    Ok(())
}
fn relation_permission_denied<T>(table: &str, kind: &str) -> Result<T, SQLError> {
    let relation =
        uqa_core::RelationIdentity::from_legacy_name(table).map_err(SQLError::Internal)?;
    Err(SQLError::Routine {
        sqlstate: "42501".into(),
        message: format!("permission denied for {kind} {}", relation.name),
    })
}