From f21adf2073b60b0cee1a0c115db53e3006a00825 Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Fri, 27 Aug 2021 14:29:28 -0700
Subject: [PATCH v6 19/19] Giving role owners control over owned roles.
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

For any set of roles (R₁..Rₓ), where Rᵢ owns Rᵢ₊₁ for i in 1..x-1,
role Rᵢ can perform any action on objects owned by role Rⱼ if i < j.

In other words, create a role ownership hierarchy where roles can
act as owners of objects belonging to other roles, so long as those
other roles have a chain of ownership leading back to the role
performing the action.
---
 src/backend/catalog/aclchk.c                  | 68 ++++++++++++++++---
 src/backend/catalog/objectaddress.c           | 22 +-----
 src/backend/commands/user.c                   | 16 +++--
 src/backend/utils/adt/acl.c                   |  4 ++
 src/include/utils/acl.h                       |  1 +
 .../expected/dummy_seclabel.out               |  2 +-
 6 files changed, 77 insertions(+), 36 deletions(-)

diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c
index 89792b154e..f24b664970 100644
--- a/src/backend/catalog/aclchk.c
+++ b/src/backend/catalog/aclchk.c
@@ -3418,6 +3418,9 @@ aclcheck_error(AclResult aclerr, ObjectType objtype,
 					case OBJECT_VIEW:
 						msg = gettext_noop("permission denied for view %s");
 						break;
+					case OBJECT_ROLE:
+						msg = gettext_noop("permission denied for role %s");
+						break;
 						/* these currently aren't used */
 					case OBJECT_ACCESS_METHOD:
 					case OBJECT_AMOP:
@@ -3428,7 +3431,6 @@ aclcheck_error(AclResult aclerr, ObjectType objtype,
 					case OBJECT_DEFACL:
 					case OBJECT_DOMCONSTRAINT:
 					case OBJECT_PUBLICATION_REL:
-					case OBJECT_ROLE:
 					case OBJECT_RULE:
 					case OBJECT_TABCONSTRAINT:
 					case OBJECT_TRANSFORM:
@@ -3543,6 +3545,9 @@ aclcheck_error(AclResult aclerr, ObjectType objtype,
 					case OBJECT_TSDICTIONARY:
 						msg = gettext_noop("must be owner of text search dictionary %s");
 						break;
+					case OBJECT_ROLE:
+						msg = gettext_noop("must be owner of role %s");
+						break;
 
 						/*
 						 * Special cases: For these, the error message talks
@@ -3567,7 +3572,6 @@ aclcheck_error(AclResult aclerr, ObjectType objtype,
 					case OBJECT_DEFACL:
 					case OBJECT_DOMCONSTRAINT:
 					case OBJECT_PUBLICATION_REL:
-					case OBJECT_ROLE:
 					case OBJECT_TRANSFORM:
 					case OBJECT_TSPARSER:
 					case OBJECT_TSTEMPLATE:
@@ -5428,16 +5432,62 @@ pg_statistics_object_ownercheck(Oid stat_oid, Oid roleid)
 	return has_privs_of_role(roleid, ownerId);
 }
 
+/*
+ * Check whether specified role is the direct or indirect owner of another
+ * role.
+ */
+bool
+pg_role_ownercheck(Oid role_oid, Oid roleid)
+{
+	HeapTuple	tuple;
+
+	/* Superusers bypass all permission checking. */
+	if (superuser_arg(roleid))
+		return true;
+
+	/*
+	 * Start with the owned role and traverse the ownership hierarchy upward.
+	 * We stop when we find the owner we are looking for or when we reach the
+	 * top.
+	 */
+	while (OidIsValid(role_oid))
+	{
+		Oid		owner_oid;
+
+		/* Find the owner of the current iteration's role */
+		tuple = SearchSysCache1(AUTHOID, ObjectIdGetDatum(role_oid));
+		if (!HeapTupleIsValid(tuple))
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("role with OID %u does not exist",
+							role_oid)));
+		owner_oid = ((Form_pg_authid) GETSTRUCT(tuple))->rolowner;
+		ReleaseSysCache(tuple);
+
+		/* Have we found the target role? */
+		if (roleid == owner_oid)
+			return true;
+
+		/*
+		 * If we have reached a role which owns itself, we must iterate no
+		 * further, else we fall into an infinite loop.
+		 */
+		if (role_oid == owner_oid)
+			return false;
+
+		/* Set up for the next iteration. */
+		role_oid = owner_oid;
+	}
+
+	return false;
+}
+
 /*
  * Check whether specified role has CREATEROLE privilege (or is a superuser)
  *
- * Note: roles do not have owners per se; instead we use this test in
- * places where an ownership-like permissions test is needed for a role.
- * Be sure to apply it to the role trying to do the operation, not the
- * role being operated on!	Also note that this generally should not be
- * considered enough privilege if the target role is a superuser.
- * (We don't handle that consideration here because we want to give a
- * separate error message for such cases, so the caller has to deal with it.)
+ * Note: In versions prior to PostgreSQL version 15, roles did not have owners
+ * per se; instead we used this test in places where an ownership-like
+ * permissions test was needed for a role.
  */
 bool
 has_createrole_privilege(Oid roleid)
diff --git a/src/backend/catalog/objectaddress.c b/src/backend/catalog/objectaddress.c
index deaabaeae9..de24727d2c 100644
--- a/src/backend/catalog/objectaddress.c
+++ b/src/backend/catalog/objectaddress.c
@@ -2540,25 +2540,9 @@ check_object_ownership(Oid roleid, ObjectType objtype, ObjectAddress address,
 							   NameListToString(castNode(List, object)));
 			break;
 		case OBJECT_ROLE:
-
-			/*
-			 * We treat roles as being "owned" by those with CREATEROLE priv,
-			 * except that superusers are only owned by superusers.
-			 */
-			if (superuser_arg(address.objectId))
-			{
-				if (!superuser_arg(roleid))
-					ereport(ERROR,
-							(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-							 errmsg("must be superuser")));
-			}
-			else
-			{
-				if (!has_createrole_privilege(roleid))
-					ereport(ERROR,
-							(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-							 errmsg("must have CREATEROLE privilege")));
-			}
+			if (!pg_role_ownercheck(roleid, GetUserId()))
+				aclcheck_error(ACLCHECK_NOT_OWNER, objtype,
+							   NameListToString(castNode(List, object)));
 			break;
 		case OBJECT_TSPARSER:
 		case OBJECT_TSTEMPLATE:
diff --git a/src/backend/commands/user.c b/src/backend/commands/user.c
index 815c7095ec..902c0473de 100644
--- a/src/backend/commands/user.c
+++ b/src/backend/commands/user.c
@@ -693,7 +693,7 @@ AlterRole(ParseState *pstate, AlterRoleStmt *stmt)
 			  !rolemembers &&
 			  !validUntil &&
 			  dpassword &&
-			  roleid == GetUserId()))
+			  !pg_role_ownercheck(roleid, GetUserId())))
 			ereport(ERROR,
 					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 					 errmsg("permission denied")));
@@ -894,7 +894,8 @@ AlterRoleSet(AlterRoleSetStmt *stmt)
 		}
 		else
 		{
-			if (!have_createrole_privilege() && roleid != GetUserId())
+			if (!have_createrole_privilege() &&
+				!pg_role_ownercheck(roleid, GetUserId()))
 				ereport(ERROR,
 						(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 						 errmsg("permission denied")));
@@ -946,11 +947,6 @@ DropRole(DropRoleStmt *stmt)
 				pg_auth_members_rel;
 	ListCell   *item;
 
-	if (!have_createrole_privilege())
-		ereport(ERROR,
-				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-				 errmsg("permission denied to drop role")));
-
 	/*
 	 * Scan the pg_authid relation to find the Oid of the role(s) to be
 	 * deleted.
@@ -1022,6 +1018,12 @@ DropRole(DropRoleStmt *stmt)
 					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 					 errmsg("must be superuser to drop superusers")));
 
+		if (!have_createrole_privilege() &&
+			!pg_role_ownercheck(roleid, GetUserId()))
+			ereport(ERROR,
+					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+					 errmsg("permission denied to drop role")));
+
 		/* DROP hook for the role being removed */
 		InvokeObjectDropHook(AuthIdRelationId, roleid, 0);
 
diff --git a/src/backend/utils/adt/acl.c b/src/backend/utils/adt/acl.c
index 67f8b29434..04eae9d4e5 100644
--- a/src/backend/utils/adt/acl.c
+++ b/src/backend/utils/adt/acl.c
@@ -4850,6 +4850,10 @@ has_privs_of_role(Oid member, Oid role)
 	if (superuser_arg(member))
 		return true;
 
+	/* Owners of roles have every privilge the owned role has */
+	if (pg_role_ownercheck(role, member))
+		return true;
+
 	/*
 	 * Find all the roles that member has the privileges of, including
 	 * multi-level recursion, then see if target role is any one of them.
diff --git a/src/include/utils/acl.h b/src/include/utils/acl.h
index af771c901d..ec9d480d67 100644
--- a/src/include/utils/acl.h
+++ b/src/include/utils/acl.h
@@ -316,6 +316,7 @@ extern bool pg_extension_ownercheck(Oid ext_oid, Oid roleid);
 extern bool pg_publication_ownercheck(Oid pub_oid, Oid roleid);
 extern bool pg_subscription_ownercheck(Oid sub_oid, Oid roleid);
 extern bool pg_statistics_object_ownercheck(Oid stat_oid, Oid roleid);
+extern bool pg_role_ownercheck(Oid role_oid, Oid roleid);
 extern bool has_createrole_privilege(Oid roleid);
 extern bool has_bypassrls_privilege(Oid roleid);
 
diff --git a/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out b/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out
index b2d898a7d1..14cfcd7bc0 100644
--- a/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out
+++ b/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out
@@ -55,7 +55,7 @@ SECURITY LABEL ON ROLE regress_dummy_seclabel_user3 IS 'unclassified';	-- fail (
 ERROR:  role "regress_dummy_seclabel_user3" does not exist
 SET SESSION AUTHORIZATION regress_dummy_seclabel_user2;
 SECURITY LABEL ON ROLE regress_dummy_seclabel_user2 IS 'unclassified';	-- fail (not privileged)
-ERROR:  must have CREATEROLE privilege
+ERROR:  must have CREATEROLE privilege or be the owner of the role
 RESET SESSION AUTHORIZATION;
 --
 -- Test for various types of object
-- 
2.21.1 (Apple Git-122.3)

