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        return Ok(());
30    }
31    // `get_relkind_objtype` names a composite relation's owner check by its type.
32    let kind = if kind == "composite type" {
33        "type"
34    } else {
35        kind
36    };
37    Err(SQLError::Routine {
38        sqlstate: "42501".into(),
39        message: format!("must be owner of {kind} {name}"),
40    })
41}
42
43/// `PostgreSQL` protects pinned catalog relations and TOAST relations after checking ownership. Unpinned catalog views retain ordinary relation validation.
44pub fn reject_system_relation_alter(relation: &uqa_core::RelationIdentity) -> Result<(), SQLError> {
45    let pinned = crate::catalog::SystemRelation::at(&relation.schema, &relation.name)
46        .is_some_and(|system| system.oid() < 12_000);
47    if pinned || relation.schema == "pg_toast" {
48        Err(SQLError::Routine {
49            sqlstate: "42501".into(),
50            message: format!(
51                "permission denied: \"{}\" is a system catalog",
52                relation.name
53            ),
54        })
55    } else {
56        Ok(())
57    }
58}
59
60#[cfg(test)]
61mod tests;
62
63pub struct OwnerChangeAuthority<'a> {
64    pub roles: &'a BTreeMap<String, RoleDefinition>,
65    pub memberships: &'a BTreeMap<RoleMembershipKey, RoleMembership>,
66    pub current_user: &'a dyn RoleSubject,
67    pub new_owner: &'a str,
68}
69
70impl OwnerChangeAuthority<'_> {
71    pub fn is_superuser(&self) -> bool {
72        self.current_user
73            .role_definition(self.roles)
74            .is_some_and(|role| role.has(RoleAttribute::Superuser))
75    }
76
77    pub fn require_owner_change(
78        &self,
79        owner: &str,
80        kind: &str,
81        name: &str,
82    ) -> Result<(), SQLError> {
83        require_relation_ownership(
84            name,
85            kind,
86            role_inherits(self.roles, self.memberships, self.current_user, owner),
87        )?;
88        require_set_role(
89            self.roles,
90            self.memberships,
91            self.current_user,
92            self.new_owner,
93        )
94    }
95
96    pub fn require_schema_create(
97        &self,
98        schemas: &dyn RelationOwnerSchemas,
99        schema: &str,
100    ) -> Result<(), SQLError> {
101        if self.is_superuser() {
102            return Ok(());
103        }
104        let security = schemas
105            .schema_security(schema)
106            .ok_or_else(|| SQLError::Routine {
107                sqlstate: "3F000".into(),
108                message: format!("schema \"{schema}\" does not exist"),
109            })?;
110        if super::schema::role_has_schema_privilege(
111            &security.resolve(self.roles).map_err(SQLError::Internal)?,
112            self.new_owner,
113            super::schema::SchemaAclPrivilege::Create,
114            self.roles,
115            self.memberships,
116        ) {
117            Ok(())
118        } else {
119            Err(SQLError::Routine {
120                sqlstate: "42501".into(),
121                message: format!("permission denied for schema {schema}"),
122            })
123        }
124    }
125
126    pub fn require_database_create(
127        &self,
128        security: &super::database::BoundDatabaseSecurity,
129    ) -> Result<(), SQLError> {
130        if super::database::role_has_database_privilege(
131            &security.resolve(self.roles).map_err(SQLError::Internal)?,
132            self.current_user,
133            super::database::DatabaseAclPrivilege::Create,
134            self.roles,
135            self.memberships,
136        ) {
137            Ok(())
138        } else {
139            Err(SQLError::Routine {
140                sqlstate: "42501".into(),
141                message: format!(
142                    "permission denied for database {}",
143                    crate::catalog::DATABASE_NAME
144                ),
145            })
146        }
147    }
148}
uqa_sql/catalog/security/ownership.rs

uqa_sql/catalog/security/
ownership.rs
