uqa_sql/catalog/security/table/
invariants.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Validate table and column ACL grant paths against durable relation metadata.
8use super::{RoleDefinition, TableAclEntry, TableAclPrivilege, TableSecurity};
9use crate::catalog::security::columns::column_grant_option_roles;
10use std::collections::{BTreeMap, BTreeSet};
11
12pub fn validate_table_security_invariants(
13    security: &TableSecurity,
14    columns: Option<&[String]>,
15    roles: &BTreeMap<String, RoleDefinition>,
16) -> Result<(), String> {
17    let validate_acl = |acl: &[TableAclEntry], column: Option<&str>| -> Result<(), String> {
18        let mut paths = BTreeSet::new();
19        for entry in acl {
20            let grantor = super::acl_grantor(entry, &security.role_owner);
21            if entry
22                .role
23                .role_name()
24                .is_some_and(|name| !roles.contains_key(name))
25            {
26                return Err(format!(
27                    "ACL references missing grantee role `{}`",
28                    entry.role
29                ));
30            }
31            if !roles.contains_key(grantor) {
32                return Err(format!("ACL references missing grantor role `{grantor}`"));
33            }
34            if !paths.insert((&entry.role, grantor)) {
35                return Err(format!(
36                    "ACL contains duplicate grant path `{grantor}` -> `{}`",
37                    entry.role
38                ));
39            }
40            if entry.privileges.is_empty() && entry.grant_options.is_empty() {
41                return Err("ACL contains an empty grant path".into());
42            }
43            if entry.role.is_public() && !entry.grant_options.is_empty() {
44                return Err("PUBLIC cannot hold grant options".into());
45            }
46            for privilege in TableAclPrivilege::ALL {
47                let mask = privilege.mask();
48                if entry.grant_options.intersects(mask) && !entry.privileges.intersects(mask) {
49                    return Err("ACL grant option exists without its privilege".into());
50                }
51                if entry.privileges.intersects(mask) || entry.grant_options.intersects(mask) {
52                    let reachable = column.map_or_else(
53                        || super::grant_option_roles(security, privilege),
54                        |column| column_grant_option_roles(security, column, privilege),
55                    );
56                    if !reachable.contains(grantor) {
57                        return Err(format!(
58                            "ACL grant path from `{grantor}` is not rooted at owner `{}`",
59                            security.role_owner
60                        ));
61                    }
62                }
63            }
64        }
65        Ok(())
66    };
67
68    if let Some(acl) = security.acl.as_deref() {
69        validate_acl(acl, None)?;
70    }
71    if !security.column_acls.is_empty() && columns.is_none() {
72        return Err("column ACLs require durable public column metadata".into());
73    }
74    for (column, acl) in &security.column_acls {
75        if !columns.is_some_and(|columns| columns.iter().any(|candidate| candidate == column)) {
76            return Err(format!("column ACL references missing column `{column}`"));
77        }
78        for entry in acl {
79            if entry.privileges.delete
80                || entry.privileges.truncate
81                || entry.privileges.trigger
82                || entry.privileges.maintain
83                || entry.grant_options.delete
84                || entry.grant_options.truncate
85                || entry.grant_options.trigger
86                || entry.grant_options.maintain
87            {
88                return Err(format!(
89                    "column ACL for `{column}` contains a relation-only privilege"
90                ));
91            }
92        }
93        validate_acl(acl, Some(column))?;
94    }
95    Ok(())
96}
uqa_sql/catalog/security/table/invariants.rs

uqa_sql/catalog/security/table/
invariants.rs
