uqa_sql/catalog/security/sequence/
invariants.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Validate sequence ACL endpoints and grant paths before catalog publication or migration.
8
9use super::{AclPrivilege, BTreeMap, BTreeSet, RoleDefinition, SequenceSecurity};
10
11pub fn validate_sequence_security_invariants(
12    security: &SequenceSecurity,
13    roles: &BTreeMap<String, RoleDefinition>,
14) -> Result<(), String> {
15    if !roles.contains_key(&security.role_owner) {
16        return Err(format!(
17            "sequence references missing owner role `{}`",
18            security.role_owner
19        ));
20    }
21    let mut paths = BTreeSet::new();
22    for entry in security.acl.iter().flatten() {
23        let grantor = super::acl_grantor(entry, &security.role_owner);
24        if entry
25            .role
26            .role_name()
27            .is_some_and(|name| !roles.contains_key(name))
28        {
29            return Err(format!(
30                "ACL references missing grantee role `{}`",
31                entry.role
32            ));
33        }
34        if !roles.contains_key(grantor) {
35            return Err(format!("ACL references missing grantor role `{grantor}`"));
36        }
37        if !paths.insert((&entry.role, grantor)) {
38            return Err(format!(
39                "ACL contains duplicate grant path `{grantor}` -> `{}`",
40                entry.role
41            ));
42        }
43        if entry.privileges.is_empty() && entry.grant_options.is_empty() {
44            return Err("ACL contains an empty grant path".into());
45        }
46        if entry.role.is_public() && !entry.grant_options.is_empty() {
47            return Err("PUBLIC cannot hold grant options".into());
48        }
49        for privilege in [
50            AclPrivilege::Select,
51            AclPrivilege::Update,
52            AclPrivilege::Usage,
53        ] {
54            let mask = privilege.mask();
55            if entry.grant_options.intersects(mask) && !entry.privileges.intersects(mask) {
56                return Err("ACL grant option exists without its privilege".into());
57            }
58            if (entry.privileges.intersects(mask) || entry.grant_options.intersects(mask))
59                && !super::grant_option_roles(security, privilege).contains(grantor)
60            {
61                return Err(format!(
62                    "ACL grant path from `{grantor}` is not rooted at owner `{}`",
63                    security.role_owner
64                ));
65            }
66        }
67    }
68    Ok(())
69}
uqa_sql/catalog/security/sequence/invariants.rs

uqa_sql/catalog/security/sequence/
invariants.rs
