Some fixes for db_owner role (#3355) (#3369)

This commit addresses the following issues in the implementation of
supporting adding of multiple users to db_owner role:
 - Name clash around internal db_owner role: We will now check if the
   internal role we are about to create already exists or not, if it
   does, we throw an error. Conversely, if a server/database principal
   exists with the same name as the internal role we are about to
   create, we will throw an appropriate error.
 - Extra rows for sp_helpuser: Added a check for RoleName column to
   only contain T-SQL roles.
 - Unable to make db_owner role schema owner: We will not change owner
   of schema if it's authorization lies with db_owner role.
 - Adding/dropping of valid users to db_owner role again should not
   throw error: During addition, if user is already member of db_owner,
   we will omit creating internal role. During drop, if user is not
   member of db_owner role, we will omit dropping the internal role.

Task: BABEL-4899, BABEL-5491, BABEL-5502

Signed-off-by: Sharu Goel <[email protected]>

contrib/babelfishpg_tsql/sql/babelfishpg_tsql.sql

@@ -2229,6 +2229,7 @@ BEGIN
 		LEFT OUTER JOIN pg_catalog.pg_roles AS Base4 ON Base4.rolname = Bsdb.owner
 		WHERE Ext1.database_name = DB_NAME()
 		AND Ext1.type != 'R'
+		AND ((Ext2.orig_username IS NULL AND Base2.oid IS NULL) OR Ext2.type = 'R') -- We should only show public if user has no members i.e. Base2.oid is NULL
 		AND Ext1.orig_username NOT IN ('db_owner', 'db_securityadmin', 'db_accessadmin', 'db_datareader', 'db_datawriter', 'db_ddladmin')
 		ORDER BY UserName, RoleName;
 	END
@@ -2299,6 +2300,7 @@ BEGIN
 		LEFT OUTER JOIN pg_catalog.pg_roles AS Base4 ON Base4.rolname = Bsdb.owner
 		WHERE Ext1.database_name = DB_NAME()
 		AND Ext1.type != 'R'
+		AND ((Ext2.orig_username IS NULL AND Base2.oid IS NULL) OR Ext2.type = 'R') -- We should only show public if user has no members i.e. Base2.oid is NULL
 		AND Ext1.orig_username NOT IN ('db_owner', 'db_securityadmin', 'db_accessadmin', 'db_datareader', 'db_datawriter', 'db_ddladmin')
 		AND (Ext1.orig_username = @name_in_db OR pg_catalog.lower(Ext1.orig_username) = pg_catalog.lower(@name_in_db))
 		ORDER BY UserName, RoleName;

contrib/babelfishpg_tsql/sql/upgrades/babelfishpg_tsql--4.4.0--5.0.0.sql

@@ -276,7 +276,7 @@ GRANT SELECT ON information_schema_tsql.key_column_usage TO PUBLIC;
 UPDATE sys.babelfish_authid_login_ext SET is_fixed_role = 1 WHERE rolname = 'sysadmin';
 
 -- At this point, there will be only one database fixed role which is db_owner.
-UPDATE sys.babelfish_authid_user_ext SET is_fixed_role = 1 WHERE orig_username = 'db_owner';
+UPDATE sys.babelfish_authid_user_ext SET is_fixed_role = 1, type = 'R' WHERE orig_username = 'db_owner';
 UPDATE sys.babelfish_authid_user_ext SET is_fixed_role = 0 WHERE orig_username != 'db_owner';
 
 CREATE OR REPLACE PROCEDURE sys.create_db_roles_during_upgrade()
@@ -661,6 +661,7 @@ BEGIN
 		LEFT OUTER JOIN pg_catalog.pg_roles AS Base4 ON Base4.rolname = Bsdb.owner
 		WHERE Ext1.database_name = DB_NAME()
 		AND Ext1.type != 'R'
+		AND ((Ext2.orig_username IS NULL AND Base2.oid IS NULL) OR Ext2.type = 'R') -- We should only show public if user has no members i.e. Base2.oid is NULL
 		AND Ext1.orig_username NOT IN ('db_owner', 'db_securityadmin', 'db_accessadmin', 'db_datareader', 'db_datawriter', 'db_ddladmin')
 		ORDER BY UserName, RoleName;
 	END
@@ -731,6 +732,7 @@ BEGIN
 		LEFT OUTER JOIN pg_catalog.pg_roles AS Base4 ON Base4.rolname = Bsdb.owner
 		WHERE Ext1.database_name = DB_NAME()
 		AND Ext1.type != 'R'
+		AND ((Ext2.orig_username IS NULL AND Base2.oid IS NULL) OR Ext2.type = 'R') -- We should only show public if user has no members i.e. Base2.oid is NULL
 		AND Ext1.orig_username NOT IN ('db_owner', 'db_securityadmin', 'db_accessadmin', 'db_datareader', 'db_datawriter', 'db_ddladmin')
 		AND (Ext1.orig_username = @name_in_db OR pg_catalog.lower(Ext1.orig_username) = pg_catalog.lower(@name_in_db))
 		ORDER BY UserName, RoleName;

contrib/babelfishpg_tsql/src/pl_handler.c

@@ -2737,9 +2737,36 @@ bbf_ProcessUtility(PlannedStmt *pstmt,
 					bool		isrole = false;
 					bool		from_windows = false;
 					Oid 		save_userid;
+					Oid 		role_oid = InvalidOid;
 					int 		save_sec_context;
 					const char	*old_createrole_self_grant;
 
+					/* Throw error if there is a possibility of name clash with an internal role */
+					role_oid = get_role_oid(stmt->role, true);
+
+					if (OidIsValid(role_oid) &&
+						(is_admin_of_role(get_bbf_role_admin_oid(), role_oid)))
+					{
+						HeapTuple tuple_cache = SearchSysCache1(AUTHIDUSEREXTROLENAME, CStringGetDatum(stmt->role));
+
+						/*
+						 * If role:
+						 *  - Is a valid PG role
+						 *  - Does not exist in Babelfish catalogs
+						 *  - bbf_role_admin is admin of this role
+						 *
+						 * We can safely assume it is a role created internally by us
+						 */
+						if (!HeapTupleIsValid(tuple_cache) && !is_login_name(stmt->role))
+							ereport(ERROR,
+								(errcode(ERRCODE_DUPLICATE_OBJECT),
+								 errmsg("Cannot create database principal \"%s\" as there already exists "
+										"a Babelfish internal role with the same name", stmt->role)));
+
+						if (HeapTupleIsValid(tuple_cache))
+							ReleaseSysCache(tuple_cache);
+					}
+
 					/* Check if creating login or role. Expect islogin first */
 					if (stmt->options != NIL)
 					{
@@ -3801,6 +3828,7 @@ bbf_ProcessUtility(PlannedStmt *pstmt,
 					else if (rolspec && strcmp(queryString, CREATE_FIXED_DB_ROLES) != 0)
 					{
 						const char *db_name = get_current_pltsql_db_name();
+						Oid db_owner_oid = InvalidOid;
 
 						owner_oid = get_rolespec_oid(rolspec, true);
 						/*
@@ -3818,7 +3846,10 @@ bbf_ProcessUtility(PlannedStmt *pstmt,
 							alter_owner = true;
 						}
 
-						if (has_privs_of_role(owner_oid, get_db_owner_oid(db_name, false)) && owner_oid != get_dbo_oid(db_name, false))
+						db_owner_oid = get_db_owner_oid(db_name, false);
+
+						if (has_privs_of_role(owner_oid, db_owner_oid) &&
+							owner_oid != get_dbo_oid(db_name, false) && owner_oid != db_owner_oid)
 						{
 							const char* new_owner = get_obj_role(get_rolespec_name(rolspec));
 							create_schema->authrole = make_rolespec_node(new_owner);

contrib/babelfishpg_tsql/src/rolecmds.c

@@ -2826,7 +2826,7 @@ get_obj_role(const char *rolname)
 	initStringInfo(&rolname_obj);
 
 	appendStringInfoString(&rolname_obj, rolname);
-	appendStringInfoString(&rolname_obj, "_obj");
+	appendStringInfoString(&rolname_obj, "_bbfobj");
 
 	truncate_tsql_identifier(rolname_obj.data);
 
@@ -2840,7 +2840,7 @@ static List
 *gen_alter_dbowner_add_subcmds(const char *rolname, const char* dbname)
 {
 	StringInfoData	query;
-	List		*stmt_list;
+	List		*stmt_list = NIL;
 	Node		*stmt;
 	int		expected_stmts = 7;
 	int		i = 0;
@@ -2852,10 +2852,21 @@ static List
 	ScanKeyData	skey;
 	Relation	rel;
 
+	/* If role is already member of db_owner role, do nothing */
+	if (has_privs_of_role(get_role_oid(rolname, false), get_db_owner_oid(dbname, false)))
+		return stmt_list;
+
 	initStringInfo(&query);
 
 	truncate_tsql_identifier(rolname_obj);
 
+	/* Throw relevant error if there will be a name clash */
+	if (get_role_oid(rolname_obj, true) != InvalidOid)
+		ereport(ERROR,
+				(errcode(ERRCODE_DUPLICATE_OBJECT),
+				 errmsg("Internal role \"%s\" could not be created because a "
+						"role already exists with the same name", rolname_obj)));
+
 	appendStringInfoString(&query, "CREATE ROLE dummy; ");
 	appendStringInfoString(&query, "GRANT dummy TO dummy; ");
 
@@ -2940,7 +2951,7 @@ static List
 *gen_alter_dbowner_drop_subcmds(const char *rolname, const char* dbname)
 {
 	StringInfoData query;
-	List		*stmt_list;
+	List		*stmt_list = NIL;
 	Node		*stmt;
 	int			expected_stmts = 7;
 	int			i = 0;
@@ -2952,6 +2963,10 @@ static List
 	ScanKeyData skey;
 	Relation	rel;
 
+	/* If role is already not a member of db_owner role, do nothing */
+	if (!has_privs_of_role(get_role_oid(rolname, false), get_db_owner_oid(dbname, false)))
+		return stmt_list;
+
 	initStringInfo(&query);
 
 	appendStringInfoString(&query, "REVOKE dummy FROM dummy; ");
@@ -3098,6 +3113,7 @@ void
 change_object_owner_if_db_owner()
 {
 	Oid		dbo_id = InvalidOid;
+	Oid		db_owner_id = InvalidOid;
 	StringInfoData	query;
 	char		*rolname = NULL;
 	char 		*obj_rolname = NULL;
@@ -3113,9 +3129,11 @@ change_object_owner_if_db_owner()
 
 	cur_db_name = get_cur_db_name();
 	dbo_id = get_dbo_oid(cur_db_name, true);
+	db_owner_id = get_db_owner_oid(cur_db_name, true);
 
-	/* Don't change object owner if current user is dbo */
-	if (role_oid == dbo_id || dbo_id == InvalidOid)
+	/* Don't change object owner if database principal is dbo or db_owner */
+	if (role_oid == dbo_id || dbo_id == InvalidOid ||
+		role_oid == db_owner_id || db_owner_id == InvalidOid)
 		return;
 
 	rolname = GetUserNameFromId(role_oid, true);
@@ -3126,7 +3144,7 @@ change_object_owner_if_db_owner()
 	if (!user_exists_for_db(cur_db_name, rolname))
 		return;
 
-	if (!is_member_of_role(role_oid, get_db_owner_oid(cur_db_name, false)))
+	if (!is_member_of_role(role_oid, db_owner_id))
 		return;
 
 	obj_rolname = get_obj_role(rolname);
@@ -3176,7 +3194,7 @@ PG_FUNCTION_INFO_V1(bbf_is_role_member);
 /*
  * The bbf_is_role_memeber function will check if a user (u1) is member of a role (r1):
  * - If role is not a member of db_owner, we will check if u1 is member of r1 role
- * - If role is a member of db_owner, we will check if u1 is member of r1_obj role
+ * - If role is a member of db_owner, we will check if u1 is member of r1_bbfobj role
  */
 Datum
 bbf_is_role_member(PG_FUNCTION_ARGS)
@@ -3212,7 +3230,7 @@ bbf_is_role_member(PG_FUNCTION_ARGS)
 /*
  * This helper function will first check if user is member of
  * db_owner role. If it is, then we will attempt drop the linked
- * "_obj" role. If it is not, this function will do nothing.
+ * "_bbfobj" role. If it is not, this function will do nothing.
  */
 static void
 drop_db_owner_related_roles(Oid roleid, const char* rolname)