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.role != "PUBLIC" && !roles.contains_key(&entry.role) {
22                return Err(format!(
23                    "ACL references missing grantee role `{}`",
24                    entry.role
25                ));
26            }
27            if grantor == "PUBLIC" || !roles.contains_key(grantor) {
28                return Err(format!("ACL references missing grantor role `{grantor}`"));
29            }
30            if !paths.insert((entry.role.as_str(), grantor)) {
31                return Err(format!(
32                    "ACL contains duplicate grant path `{grantor}` -> `{}`",
33                    entry.role
34                ));
35            }
36            if entry.privileges.is_empty() && entry.grant_options.is_empty() {
37                return Err("ACL contains an empty grant path".into());
38            }
39            if entry.role == "PUBLIC" && !entry.grant_options.is_empty() {
40                return Err("PUBLIC cannot hold grant options".into());
41            }
42            for privilege in TableAclPrivilege::ALL {
43                let mask = privilege.mask();
44                if entry.grant_options.intersects(mask) && !entry.privileges.intersects(mask) {
45                    return Err("ACL grant option exists without its privilege".into());
46                }
47                if entry.privileges.intersects(mask) || entry.grant_options.intersects(mask) {
48                    let reachable = column.map_or_else(
49                        || super::grant_option_roles(security, privilege),
50                        |column| column_grant_option_roles(security, column, privilege),
51                    );
52                    if !reachable.contains(grantor) {
53                        return Err(format!(
54                            "ACL grant path from `{grantor}` is not rooted at owner `{}`",
55                            security.role_owner
56                        ));
57                    }
58                }
59            }
60        }
61        Ok(())
62    };
63
64    if let Some(acl) = security.acl.as_deref() {
65        validate_acl(acl, None)?;
66    }
67    if !security.column_acls.is_empty() && columns.is_none() {
68        return Err("column ACLs require durable public column metadata".into());
69    }
70    for (column, acl) in &security.column_acls {
71        if !columns.is_some_and(|columns| columns.iter().any(|candidate| candidate == column)) {
72            return Err(format!("column ACL references missing column `{column}`"));
73        }
74        for entry in acl {
75            if entry.privileges.delete
76                || entry.privileges.truncate
77                || entry.privileges.trigger
78                || entry.privileges.maintain
79                || entry.grant_options.delete
80                || entry.grant_options.truncate
81                || entry.grant_options.trigger
82                || entry.grant_options.maintain
83            {
84                return Err(format!(
85                    "column ACL for `{column}` contains a relation-only privilege"
86                ));
87            }
88        }
89        validate_acl(acl, Some(column))?;
90    }
91    Ok(())
92}
uqa_sql/catalog/security/table/invariants.rs

uqa_sql/catalog/security/table/
invariants.rs
