uqa_sql/catalog/security/
ownership.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Read-only authority shared by object ownership checks.
8use super::BoundSchemaSecurity;
9use crate::catalog::roles::identity::RoleSubject;
10use crate::{
11    ast::RoleAttribute,
12    catalog::roles::{
13        require_set_role, role_inherits, RoleDefinition, RoleMembership, RoleMembershipKey,
14    },
15    SQLError,
16};
17use std::collections::BTreeMap;
18
19pub trait RelationOwnerSchemas {
20    fn schema_security(&self, schema: &str) -> Option<BoundSchemaSecurity>;
21}
22
23pub fn require_relation_ownership(
24    name: &str,
25    kind: &str,
26    has_owner_privileges: bool,
27) -> Result<(), SQLError> {
28    if has_owner_privileges {
29        Ok(())
30    } else {
31        Err(SQLError::Routine {
32            sqlstate: "42501".into(),
33            message: format!("must be owner of {kind} {name}"),
34        })
35    }
36}
37
38/// `PostgreSQL` protects pinned catalog relations and TOAST relations after checking ownership. Unpinned catalog views retain ordinary relation validation.
39pub fn reject_system_relation_alter(relation: &uqa_core::RelationIdentity) -> Result<(), SQLError> {
40    let pinned = crate::catalog::SystemRelation::at(&relation.schema, &relation.name)
41        .is_some_and(|system| system.oid() < 12_000);
42    if pinned || relation.schema == "pg_toast" {
43        Err(SQLError::Routine {
44            sqlstate: "42501".into(),
45            message: format!(
46                "permission denied: \"{}\" is a system catalog",
47                relation.name
48            ),
49        })
50    } else {
51        Ok(())
52    }
53}
54
55#[cfg(test)]
56mod tests;
57
58pub struct OwnerChangeAuthority<'a> {
59    pub roles: &'a BTreeMap<String, RoleDefinition>,
60    pub memberships: &'a BTreeMap<RoleMembershipKey, RoleMembership>,
61    pub current_user: &'a dyn RoleSubject,
62    pub new_owner: &'a str,
63}
64
65impl OwnerChangeAuthority<'_> {
66    pub fn is_superuser(&self) -> bool {
67        self.current_user
68            .role_definition(self.roles)
69            .is_some_and(|role| role.has(RoleAttribute::Superuser))
70    }
71
72    pub fn require_owner_change(
73        &self,
74        owner: &str,
75        kind: &str,
76        name: &str,
77    ) -> Result<(), SQLError> {
78        require_relation_ownership(
79            name,
80            kind,
81            role_inherits(self.roles, self.memberships, self.current_user, owner),
82        )?;
83        require_set_role(
84            self.roles,
85            self.memberships,
86            self.current_user,
87            self.new_owner,
88        )
89    }
90
91    pub fn require_schema_create(
92        &self,
93        schemas: &dyn RelationOwnerSchemas,
94        schema: &str,
95    ) -> Result<(), SQLError> {
96        if self.is_superuser() {
97            return Ok(());
98        }
99        let security = schemas
100            .schema_security(schema)
101            .ok_or_else(|| SQLError::Routine {
102                sqlstate: "3F000".into(),
103                message: format!("schema \"{schema}\" does not exist"),
104            })?;
105        if super::schema::role_has_schema_privilege(
106            &security.resolve(self.roles).map_err(SQLError::Internal)?,
107            self.new_owner,
108            super::schema::SchemaAclPrivilege::Create,
109            self.roles,
110            self.memberships,
111        ) {
112            Ok(())
113        } else {
114            Err(SQLError::Routine {
115                sqlstate: "42501".into(),
116                message: format!("permission denied for schema {schema}"),
117            })
118        }
119    }
120
121    pub fn require_database_create(
122        &self,
123        security: &super::database::BoundDatabaseSecurity,
124    ) -> Result<(), SQLError> {
125        if super::database::role_has_database_privilege(
126            &security.resolve(self.roles).map_err(SQLError::Internal)?,
127            self.current_user,
128            super::database::DatabaseAclPrivilege::Create,
129            self.roles,
130            self.memberships,
131        ) {
132            Ok(())
133        } else {
134            Err(SQLError::Routine {
135                sqlstate: "42501".into(),
136                message: format!(
137                    "permission denied for database {}",
138                    crate::catalog::DATABASE_NAME
139                ),
140            })
141        }
142    }
143}
uqa_sql/catalog/security/ownership.rs

uqa_sql/catalog/security/
ownership.rs
