Add "pg_database_owner" default role.

Membership consists, implicitly, of the current database owner.  Expect
use in template databases.  Once pg_database_owner has rights within a
template, each owner of a database instantiated from that template will
exercise those rights.

Reviewed by John Naylor.

Discussion: https://postgr.es/m/20201228043148.GA1053024@rfd.leadboat.com

src/backend/catalog/information_schema.sql

@@ -255,7 +255,14 @@ CREATE VIEW applicable_roles AS
     SELECT CAST(a.rolname AS sql_identifier) AS grantee,
            CAST(b.rolname AS sql_identifier) AS role_name,
            CAST(CASE WHEN m.admin_option THEN 'YES' ELSE 'NO' END AS yes_or_no) AS is_grantable
-    FROM pg_auth_members m
+    FROM (SELECT member, roleid, admin_option FROM pg_auth_members
+          -- This UNION could be UNION ALL, but UNION works even if we start
+          -- to allow explicit pg_database_owner membership.
+          UNION
+          SELECT datdba, pg_authid.oid, false
+          FROM pg_database, pg_authid
+          WHERE datname = current_database() AND rolname = 'pg_database_owner'
+         )  m
          JOIN pg_authid a ON (m.member = a.oid)
          JOIN pg_authid b ON (m.roleid = b.oid)
     WHERE pg_has_role(a.oid, 'USAGE');

src/backend/commands/user.c

@@ -1496,6 +1496,18 @@ AddRoleMems(const char *rolename, Oid roleid,
 							rolename)));
 	}
 
+	/*
+	 * The charter of pg_database_owner is to have exactly one, implicit,
+	 * situation-dependent member.  There's no technical need for this
+	 * restriction.  (One could lift it and take the further step of making
+	 * pg_database_ownercheck() equivalent to has_privs_of_role(roleid,
+	 * DEFAULT_ROLE_DATABASE_OWNER), in which case explicit,
+	 * situation-independent members could act as the owner of any database.)
+	 */
+	if (roleid == DEFAULT_ROLE_DATABASE_OWNER)
+		ereport(ERROR,
+				errmsg("role \"%s\" cannot have explicit members", rolename));
+
 	/*
 	 * The role membership grantor of record has little significance at
 	 * present.  Nonetheless, inasmuch as users might look to it for a crude
@@ -1524,6 +1536,30 @@ AddRoleMems(const char *rolename, Oid roleid,
 		bool		new_record_nulls[Natts_pg_auth_members];
 		bool		new_record_repl[Natts_pg_auth_members];
 
+		/*
+		 * pg_database_owner is never a role member.  Lifting this restriction
+		 * would require a policy decision about membership loops.  One could
+		 * prevent loops, which would include making "ALTER DATABASE x OWNER
+		 * TO proposed_datdba" fail if is_member_of_role(pg_database_owner,
+		 * proposed_datdba).  Hence, gaining a membership could reduce what a
+		 * role could do.  Alternately, one could allow these memberships to
+		 * complete loops.  A role could then have actual WITH ADMIN OPTION on
+		 * itself, prompting a decision about is_admin_of_role() treatment of
+		 * the case.
+		 *
+		 * Lifting this restriction also has policy implications for ownership
+		 * of shared objects (databases and tablespaces).  We allow such
+		 * ownership, but we might find cause to ban it in the future.
+		 * Designing such a ban would more troublesome if the design had to
+		 * address pg_database_owner being a member of role FOO that owns a
+		 * shared object.  (The effect of such ownership is that any owner of
+		 * another database can act as the owner of affected shared objects.)
+		 */
+		if (memberid == DEFAULT_ROLE_DATABASE_OWNER)
+			ereport(ERROR,
+					errmsg("role \"%s\" cannot be a member of any role",
+						   get_rolespec_name(memberRole)));
+
 		/*
 		 * Refuse creation of membership loops, including the trivial case
 		 * where a role is made a member of itself.  We do this by checking to

src/backend/utils/adt/acl.c

@@ -22,6 +22,7 @@
 #include "catalog/pg_auth_members.h"
 #include "catalog/pg_authid.h"
 #include "catalog/pg_class.h"
+#include "catalog/pg_database.h"
 #include "catalog/pg_type.h"
 #include "commands/dbcommands.h"
 #include "commands/proclang.h"
@@ -68,6 +69,7 @@ enum RoleRecurseType
 };
 static Oid	cached_role[] = {InvalidOid, InvalidOid};
 static List *cached_roles[] = {NIL, NIL};
+static uint32 cached_db_hash;
 
 
 static const char *getid(const char *s, char *n);
@@ -4665,10 +4667,14 @@ initialize_acl(void)
 {
 	if (!IsBootstrapProcessingMode())
 	{
+		cached_db_hash =
+			GetSysCacheHashValue1(DATABASEOID,
+								  ObjectIdGetDatum(MyDatabaseId));
+
 		/*
 		 * In normal mode, set a callback on any syscache invalidation of rows
-		 * of pg_auth_members (for roles_is_member_of()) or pg_authid (for
-		 * has_rolinherit())
+		 * of pg_auth_members (for roles_is_member_of()), pg_authid (for
+		 * has_rolinherit()), or pg_database (for roles_is_member_of())
 		 */
 		CacheRegisterSyscacheCallback(AUTHMEMROLEMEM,
 									  RoleMembershipCacheCallback,
@@ -4676,6 +4682,9 @@ initialize_acl(void)
 		CacheRegisterSyscacheCallback(AUTHOID,
 									  RoleMembershipCacheCallback,
 									  (Datum) 0);
+		CacheRegisterSyscacheCallback(DATABASEOID,
+									  RoleMembershipCacheCallback,
+									  (Datum) 0);
 	}
 }
 
@@ -4686,6 +4695,13 @@ initialize_acl(void)
 static void
 RoleMembershipCacheCallback(Datum arg, int cacheid, uint32 hashvalue)
 {
+	if (cacheid == DATABASEOID &&
+		hashvalue != cached_db_hash &&
+		hashvalue != 0)
+	{
+		return;					/* ignore pg_database changes for other DBs */
+	}
+
 	/* Force membership caches to be recomputed on next use */
 	cached_role[ROLERECURSE_PRIVS] = InvalidOid;
 	cached_role[ROLERECURSE_MEMBERS] = InvalidOid;
@@ -4728,6 +4744,7 @@ static List *
 roles_is_member_of(Oid roleid, enum RoleRecurseType type,
 				   Oid admin_of, bool *is_admin)
 {
+	Oid			dba;
 	List	   *roles_list;
 	ListCell   *l;
 	List	   *new_cached_roles;
@@ -4740,6 +4757,24 @@ roles_is_member_of(Oid roleid, enum RoleRecurseType type,
 		OidIsValid(cached_role[type]))
 		return cached_roles[type];
 
+	/*
+	 * Role expansion happens in a non-database backend when guc.c checks
+	 * DEFAULT_ROLE_READ_ALL_SETTINGS for a physical walsender SHOW command.
+	 * In that case, no role gets pg_database_owner.
+	 */
+	if (!OidIsValid(MyDatabaseId))
+		dba = InvalidOid;
+	else
+	{
+		HeapTuple	dbtup;
+
+		dbtup = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(MyDatabaseId));
+		if (!HeapTupleIsValid(dbtup))
+			elog(ERROR, "cache lookup failed for database %u", MyDatabaseId);
+		dba = ((Form_pg_database) GETSTRUCT(dbtup))->datdba;
+		ReleaseSysCache(dbtup);
+	}
+
 	/*
 	 * Find all the roles that roleid is a member of, including multi-level
 	 * recursion.  The role itself will always be the first element of the
@@ -4787,6 +4822,11 @@ roles_is_member_of(Oid roleid, enum RoleRecurseType type,
 			roles_list = list_append_unique_oid(roles_list, otherid);
 		}
 		ReleaseSysCacheList(memlist);
+
+		/* implement pg_database_owner implicit membership */
+		if (memberid == dba && OidIsValid(dba))
+			roles_list = list_append_unique_oid(roles_list,
+												DEFAULT_ROLE_DATABASE_OWNER);
 	}
 
 	/*

src/backend/utils/cache/catcache.c

@@ -1076,8 +1076,9 @@ InitCatCachePhase2(CatCache *cache, bool touch_index)
  *		criticalRelcachesBuilt), we don't have to worry anymore.
  *
  *		Similarly, during backend startup we have to be able to use the
- *		pg_authid and pg_auth_members syscaches for authentication even if
- *		we don't yet have relcache entries for those catalogs' indexes.
+ *		pg_authid, pg_auth_members and pg_database syscaches for
+ *		authentication even if we don't yet have relcache entries for those
+ *		catalogs' indexes.
  */
 static bool
 IndexScanOK(CatCache *cache, ScanKey cur_skey)
@@ -1110,6 +1111,7 @@ IndexScanOK(CatCache *cache, ScanKey cur_skey)
 		case AUTHNAME:
 		case AUTHOID:
 		case AUTHMEMMEMROLE:
+		case DATABASEOID:
 
 			/*
 			 * Protect authentication lookups occurring before relcache has

src/bin/psql/describe.c

@@ -3557,6 +3557,7 @@ describeRoles(const char *pattern, bool verbose, bool showSystem)
 
 	printTableAddHeader(&cont, gettext_noop("Role name"), true, align);
 	printTableAddHeader(&cont, gettext_noop("Attributes"), true, align);
+	/* ignores implicit memberships from superuser & pg_database_owner */
 	printTableAddHeader(&cont, gettext_noop("Member of"), true, align);
 
 	if (verbose && pset.sversion >= 80200)

src/include/catalog/catversion.h

@@ -53,6 +53,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	202103264
+#define CATALOG_VERSION_NO	202103265
 
 #endif

src/include/catalog/pg_authid.dat

@@ -24,6 +24,11 @@
   rolcreaterole => 't', rolcreatedb => 't', rolcanlogin => 't',
   rolreplication => 't', rolbypassrls => 't', rolconnlimit => '-1',
   rolpassword => '_null_', rolvaliduntil => '_null_' },
+{ oid => '8778', oid_symbol => 'DEFAULT_ROLE_DATABASE_OWNER',
+  rolname => 'pg_database_owner', rolsuper => 'f', rolinherit => 't',
+  rolcreaterole => 'f', rolcreatedb => 'f', rolcanlogin => 'f',
+  rolreplication => 'f', rolbypassrls => 'f', rolconnlimit => '-1',
+  rolpassword => '_null_', rolvaliduntil => '_null_' },
 { oid => '3373', oid_symbol => 'DEFAULT_ROLE_MONITOR',
   rolname => 'pg_monitor', rolsuper => 'f', rolinherit => 't',
   rolcreaterole => 'f', rolcreatedb => 'f', rolcanlogin => 'f',

src/test/regress/expected/privileges.out

@@ -1719,6 +1719,67 @@ SELECT * FROM pg_largeobject LIMIT 0;
 SET SESSION AUTHORIZATION regress_priv_user1;
 SELECT * FROM pg_largeobject LIMIT 0;			-- to be denied
 ERROR:  permission denied for table pg_largeobject
+-- test pg_database_owner
+RESET SESSION AUTHORIZATION;
+GRANT pg_database_owner TO regress_priv_user1;
+ERROR:  role "pg_database_owner" cannot have explicit members
+GRANT regress_priv_user1 TO pg_database_owner;
+ERROR:  role "pg_database_owner" cannot be a member of any role
+CREATE TABLE datdba_only ();
+ALTER TABLE datdba_only OWNER TO pg_database_owner;
+REVOKE DELETE ON datdba_only FROM pg_database_owner;
+SELECT
+	pg_has_role('regress_priv_user1', 'pg_database_owner', 'USAGE') as priv,
+	pg_has_role('regress_priv_user1', 'pg_database_owner', 'MEMBER') as mem,
+	pg_has_role('regress_priv_user1', 'pg_database_owner',
+				'MEMBER WITH ADMIN OPTION') as admin;
+ priv | mem | admin 
+------+-----+-------
+ f    | f   | f
+(1 row)
+
+BEGIN;
+DO $$BEGIN EXECUTE format(
+	'ALTER DATABASE %I OWNER TO regress_priv_group2', current_catalog); END$$;
+SELECT
+	pg_has_role('regress_priv_user1', 'pg_database_owner', 'USAGE') as priv,
+	pg_has_role('regress_priv_user1', 'pg_database_owner', 'MEMBER') as mem,
+	pg_has_role('regress_priv_user1', 'pg_database_owner',
+				'MEMBER WITH ADMIN OPTION') as admin;
+ priv | mem | admin 
+------+-----+-------
+ t    | t   | f
+(1 row)
+
+SET SESSION AUTHORIZATION regress_priv_user1;
+TABLE information_schema.enabled_roles ORDER BY role_name COLLATE "C";
+      role_name      
+---------------------
+ pg_database_owner
+ regress_priv_group2
+ regress_priv_user1
+(3 rows)
+
+TABLE information_schema.applicable_roles ORDER BY role_name COLLATE "C";
+       grantee       |      role_name      | is_grantable 
+---------------------+---------------------+--------------
+ regress_priv_group2 | pg_database_owner   | NO
+ regress_priv_user1  | regress_priv_group2 | NO
+(2 rows)
+
+INSERT INTO datdba_only DEFAULT VALUES;
+SAVEPOINT q; DELETE FROM datdba_only; ROLLBACK TO q;
+ERROR:  permission denied for table datdba_only
+SET SESSION AUTHORIZATION regress_priv_user2;
+TABLE information_schema.enabled_roles;
+     role_name      
+--------------------
+ regress_priv_user2
+(1 row)
+
+INSERT INTO datdba_only DEFAULT VALUES;
+ERROR:  permission denied for table datdba_only
+ROLLBACK;
 -- test default ACLs
 \c -
 CREATE SCHEMA testns;

src/test/regress/sql/privileges.sql

@@ -1034,6 +1034,37 @@ SELECT * FROM pg_largeobject LIMIT 0;
 SET SESSION AUTHORIZATION regress_priv_user1;
 SELECT * FROM pg_largeobject LIMIT 0;			-- to be denied
 
+-- test pg_database_owner
+RESET SESSION AUTHORIZATION;
+GRANT pg_database_owner TO regress_priv_user1;
+GRANT regress_priv_user1 TO pg_database_owner;
+CREATE TABLE datdba_only ();
+ALTER TABLE datdba_only OWNER TO pg_database_owner;
+REVOKE DELETE ON datdba_only FROM pg_database_owner;
+SELECT
+	pg_has_role('regress_priv_user1', 'pg_database_owner', 'USAGE') as priv,
+	pg_has_role('regress_priv_user1', 'pg_database_owner', 'MEMBER') as mem,
+	pg_has_role('regress_priv_user1', 'pg_database_owner',
+				'MEMBER WITH ADMIN OPTION') as admin;
+
+BEGIN;
+DO $$BEGIN EXECUTE format(
+	'ALTER DATABASE %I OWNER TO regress_priv_group2', current_catalog); END$$;
+SELECT
+	pg_has_role('regress_priv_user1', 'pg_database_owner', 'USAGE') as priv,
+	pg_has_role('regress_priv_user1', 'pg_database_owner', 'MEMBER') as mem,
+	pg_has_role('regress_priv_user1', 'pg_database_owner',
+				'MEMBER WITH ADMIN OPTION') as admin;
+SET SESSION AUTHORIZATION regress_priv_user1;
+TABLE information_schema.enabled_roles ORDER BY role_name COLLATE "C";
+TABLE information_schema.applicable_roles ORDER BY role_name COLLATE "C";
+INSERT INTO datdba_only DEFAULT VALUES;
+SAVEPOINT q; DELETE FROM datdba_only; ROLLBACK TO q;
+SET SESSION AUTHORIZATION regress_priv_user2;
+TABLE information_schema.enabled_roles;
+INSERT INTO datdba_only DEFAULT VALUES;
+ROLLBACK;
+
 -- test default ACLs
 \c -