[Fix] Database owner login is unable to view db objects in SSMS

Currently in babelfish when we create a login (e.g. postjend_svc) and a database (e.g. test) and make postjend_svc owner of test and then connect to SSMS object explorer as login "postjend_svc" to database "test" and try to expand the "Database" field to view the objects. It does not show anything. With this change the login with be able to expand the "Database" field.

The issue arrised because when the databases are expanded in SSMS, a query is being executed which contains a cross-db query. Currently, we handle cross-db queries by switching the database and current user. This cross-db query is resulting into permission denied error due to lack of permission of the current user (which is switched internally to master_guest) to insert into a table created in "test" db.

The solution of the above issue is implemented in two parts:

1 - For the successful execution of the cross-db query, I switched the current user with session user (login) which resulted in the user having adequate permission of both the databases required for the execution of the script.

BABEL-5218 was found after implementing the above fix. This issue arrised because when a non sysadmin login tries to run cross-db dml query like INSERT. The permission check is done through the pg_catalogue side causing a permission denied error because dbo role for that database is not given to that login during the "alter authorization".

2 - To fix the above issue, I assigned dbo role to the login when it is made the owner of the database. Since the login has the dbo role, it can perform the above queries successfully.

Task: BABEL-5119, BABEL-5218

Signed-off-by: P Aswini Kumar <[email protected]>

contrib/babelfishpg_tsql/src/dbcmds.c

@@ -49,7 +49,8 @@ static List *gen_createdb_subcmds(const char *schema,
 								  const char *dbo,
 								  const char *db_owner,
 								  const char *guest,
-								  const char *guest_schema);
+								  const char *guest_schema,
+								  const char *owner);
 static List *gen_dropdb_subcmds(const char *schema,
 								const char *db_owner,
 								const char *dbo,
@@ -77,14 +78,17 @@ get_sys_babelfish_db_seq_oid()
  * Generate subcmds for CREATE DATABASE. Note 'guest' can be NULL.
  */
 static List *
-gen_createdb_subcmds(const char *schema, const char *dbo, const char *db_owner, const char *guest, const char *guest_schema)
+gen_createdb_subcmds(const char *schema, const char *dbo, const char *db_owner, const char *guest, const char *guest_schema, const char *owner)
 {
 	StringInfoData query;
 	List	   *res;
 	List	   *logins = NIL;
 	Node	   *stmt;
 	int			i = 0;
 	int			expected_stmt_num;
+	AccessPriv *acc;
+	List	   *privs = NIL;
+	RoleSpec   *role_spec;
 
 	/*
 	 * To avoid SQL injection, we generate statement parsetree with dummy
@@ -95,6 +99,7 @@ gen_createdb_subcmds(const char *schema, const char *dbo, const char *db_owner,
 	appendStringInfo(&query, "CREATE ROLE dummy CREATEROLE INHERIT; ");
 	appendStringInfo(&query, "CREATE ROLE dummy INHERIT CREATEROLE ROLE sysadmin IN ROLE dummy; ");
 	appendStringInfo(&query, "GRANT CREATE, CONNECT, TEMPORARY ON DATABASE dummy TO dummy; ");
+	appendStringInfo(&query, "GRANT dummy TO dummy; ");
 
 	if (guest)
 	{
@@ -115,9 +120,9 @@ gen_createdb_subcmds(const char *schema, const char *dbo, const char *db_owner,
 	res = raw_parser(query.data, RAW_PARSE_DEFAULT);
 
 	if (guest)
-		expected_stmt_num = list_length(logins) > 0 ? 9 : 8;
+		expected_stmt_num = list_length(logins) > 0 ? 10 : 9;
 	else
-		expected_stmt_num = 6;
+		expected_stmt_num = 7;
 
 	if (list_length(res) != expected_stmt_num)
 		ereport(ERROR,
@@ -135,6 +140,19 @@ gen_createdb_subcmds(const char *schema, const char *dbo, const char *db_owner,
 	stmt = parsetree_nth_stmt(res, i++);
 	update_GrantStmt(stmt, get_database_name(MyDatabaseId), NULL, dbo, NULL);
 
+	/* Grant dbo role to owner */
+	stmt = parsetree_nth_stmt(res, i++);
+	acc = makeNode(AccessPriv);
+	acc->priv_name = pstrdup(dbo);
+	acc->cols = NIL;
+	privs = lappend(privs, acc);
+
+	role_spec = makeNode(RoleSpec);
+	role_spec->roletype = ROLESPEC_CSTRING;
+	role_spec->location = -1;
+	role_spec->rolename = pstrdup(owner);
+	update_GrantRoleStmt(stmt, privs, list_make1(role_spec));
+
 	if (guest)
 	{
 		stmt = parsetree_nth_stmt(res, i++);
@@ -512,7 +530,7 @@ create_bbf_db_internal(const char *dbname, List *options, const char *owner, int
 	/* Advance cmd counter to make the database visible */
 	CommandCounterIncrement();
 
-	parsetree_list = gen_createdb_subcmds(dbo_scm, dbo_role, db_owner_role, guest, guest_scm);
+	parsetree_list = gen_createdb_subcmds(dbo_scm, dbo_role, db_owner_role, guest, guest_scm, owner);
 
 	GetUserIdAndSecContext(&save_userid, &save_sec_context);
 	old_createrole_self_grant = pstrdup(GetConfigOption("createrole_self_grant", false, true));

contrib/babelfishpg_tsql/src/pl_exec-2.c

@@ -28,6 +28,7 @@
 #include "parser/scansup.h"
 #include "parser/parse_oper.h"
 #include "src/include/lib/qunique.h"
+#include "rolecmds.h"
 
 /* helper function to get current T-SQL estate */
 PLtsql_execstate *get_current_tsql_estate(void);
@@ -3869,6 +3870,9 @@ static int
 exec_stmt_change_dbowner(PLtsql_execstate *estate, PLtsql_stmt_change_dbowner *stmt)
 {
 	char *new_owner_is_user;
+	Oid 		save_userid;
+	int 		save_sec_context;
+	const char *prev_db_owner = get_owner_of_db(stmt->db_name);
 
 	/* Verify target database exists. */
 	if (!DbidIsValid(get_db_id(stmt->db_name)))
@@ -3921,9 +3925,29 @@ exec_stmt_change_dbowner(PLtsql_execstate *estate, PLtsql_stmt_change_dbowner *s
 		ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 						errmsg("The proposed new database owner is already a user or aliased in the database.")));				
 	}
-
-	/* All validations done, perform the actual update */
-	update_db_owner(stmt->new_owner_name, stmt->db_name);	
+
+	/* Save the previous user to be restored after granting dbo role to the login. */
+	GetUserIdAndSecContext(&save_userid, &save_sec_context);
+
+	PG_TRY();
+	{
+		/*
+		* Set current user to bbf_role_admin for grant roles.
+		*/
+		SetUserIdAndSecContext(get_bbf_role_admin_oid(), save_sec_context | SECURITY_LOCAL_USERID_CHANGE);
+
+		/* Revoke dbo role from the previous owner */
+		revoke_dbo_from_login(prev_db_owner, stmt->db_name);
+
+		/* Grant dbo role to the new owner */
+		grant_dbo_to_login(stmt->new_owner_name, stmt->db_name);
+		update_db_owner(stmt->new_owner_name, stmt->db_name);	
+	}
+	PG_FINALLY();
+	{
+		SetUserIdAndSecContext(save_userid, save_sec_context);
+	}
+	PG_END_TRY();
 	return PLTSQL_RC_OK;
 }
 

contrib/babelfishpg_tsql/src/pltsql.h

@@ -2159,6 +2159,7 @@ extern void update_DropOwnedStmt(Node *n, List *role_list);
 extern void update_DropRoleStmt(Node *n, const char *role);
 extern void update_DropStmt(Node *n, const char *object);
 extern void update_GrantRoleStmt(Node *n, List *privs, List *roles);
+extern void update_RevokeRoleStmt(Node *n, List *privs, List *roles);
 extern void update_GrantStmt(Node *n, const char *object, const char *obj_schema, const char *grantee, const char *priv);
 extern void update_RenameStmt(Node *n, const char *old_name, const char *new_name);
 extern void update_ViewStmt(Node *n, const char *view_schema);

contrib/babelfishpg_tsql/src/pltsql_utils.c

@@ -1076,6 +1076,19 @@ update_GrantRoleStmt(Node *n, List *privs, List *roles)
 	stmt->grantee_roles = roles;
 }
 
+void
+update_RevokeRoleStmt(Node *n, List *privs, List *roles)
+{
+	GrantRoleStmt *stmt = (GrantRoleStmt *) n;
+
+	if (!IsA(stmt, GrantRoleStmt))
+		ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("query is not a RevokeRoleStmt")));
+
+	stmt->is_grant = false;
+	stmt->granted_roles = privs;
+	stmt->grantee_roles = roles;
+}
+
 void
 update_GrantStmt(Node *n, const char *object, const char *obj_schema, const char *grantee, const char *priv)
 {

contrib/babelfishpg_tsql/src/rolecmds.c

@@ -583,6 +583,136 @@ grant_guests_to_login(const char *login)
 	pfree(query.data);
 }
 
+void
+grant_dbo_to_login(const char* login, const char* db_name)
+{
+	StringInfoData query;
+	List	   *parsetree_list;
+	List	   *dbo = NIL;
+	Node	   *stmt;
+	RoleSpec   *tmp;
+	PlannedStmt *wrapper;
+	AccessPriv *acc;
+
+	const char *dbo_role_name = get_dbo_role_name(db_name);
+
+	initStringInfo(&query);
+
+	acc = makeNode(AccessPriv);
+	acc->priv_name = pstrdup(dbo_role_name);
+	acc->cols = NIL;
+	dbo = lappend(dbo, acc);
+
+	/* Build dummy GRANT statement to grant membership to login  */
+	appendStringInfo(&query, "GRANT dummy TO dummy; ");
+
+	parsetree_list = raw_parser(query.data, RAW_PARSE_DEFAULT);
+
+	if (list_length(parsetree_list) != 1)
+		ereport(ERROR,
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				 errmsg("Expected 1 statement but get %d statements after parsing",
+						list_length(parsetree_list))));
+
+	/* Update the dummy statement with real values */
+	stmt = parsetree_nth_stmt(parsetree_list, 0);
+	tmp = makeNode(RoleSpec);
+	tmp->roletype = ROLESPEC_CSTRING;
+	tmp->location = -1;
+	tmp->rolename = pstrdup(login);
+
+	update_GrantRoleStmt(stmt, dbo, list_make1(tmp));
+
+	/* Run the built query */
+	/* need to make a wrapper PlannedStmt */
+	wrapper = makeNode(PlannedStmt);
+	wrapper->commandType = CMD_UTILITY;
+	wrapper->canSetTag = false;
+	wrapper->utilityStmt = stmt;
+	wrapper->stmt_location = 0;
+	wrapper->stmt_len = 18;
+
+	/* do this step */
+	ProcessUtility(wrapper,
+				   "(CREATE DATABASE )",
+				   false,
+				   PROCESS_UTILITY_SUBCOMMAND,
+				   NULL,
+				   NULL,
+				   None_Receiver,
+				   NULL);
+
+	/* make sure later steps can see the object created here */
+	CommandCounterIncrement();
+
+	pfree(query.data);
+}
+
+void
+revoke_dbo_from_login(const char* login, const char* db_name)
+{
+	StringInfoData query;
+	List	   *parsetree_list;
+	List	   *dbo = NIL;
+	Node	   *stmt;
+	RoleSpec   *tmp;
+	PlannedStmt *wrapper;
+	AccessPriv *acc;
+
+	const char *dbo_role_name = get_dbo_role_name(db_name);
+
+	initStringInfo(&query);
+
+	acc = makeNode(AccessPriv);
+	acc->priv_name = pstrdup(dbo_role_name);
+	acc->cols = NIL;
+	dbo = lappend(dbo, acc);
+
+	/* Build dummy REVOKE statement to revoke membership from login */
+	appendStringInfo(&query, "REVOKE dummy FROM dummy; ");
+
+	parsetree_list = raw_parser(query.data, RAW_PARSE_DEFAULT);
+
+	if (list_length(parsetree_list) != 1)
+		ereport(ERROR,
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				 errmsg("Expected 1 statement but get %d statements after parsing",
+						list_length(parsetree_list))));
+
+	/* Update the dummy statement with real values */
+	stmt = parsetree_nth_stmt(parsetree_list, 0);
+	tmp = makeNode(RoleSpec);
+	tmp->roletype = ROLESPEC_CSTRING;
+	tmp->location = -1;
+	tmp->rolename = pstrdup(login);
+
+	update_RevokeRoleStmt(stmt, dbo, list_make1(tmp));
+
+	/* Run the built query */
+	/* need to make a wrapper PlannedStmt */
+	wrapper = makeNode(PlannedStmt);
+	wrapper->commandType = CMD_UTILITY;
+	wrapper->canSetTag = false;
+	wrapper->utilityStmt = stmt;
+	wrapper->stmt_location = 0;
+	wrapper->stmt_len = 18;
+
+	/* do this step */
+	ProcessUtility(wrapper,
+				   "(CREATE DATABASE )",
+				   false,
+				   PROCESS_UTILITY_SUBCOMMAND,
+				   NULL,
+				   NULL,
+				   None_Receiver,
+				   NULL);
+
+	/* make sure later steps can see the object created here */
+	CommandCounterIncrement();
+
+	pfree(query.data);
+}
+
 static List *
 gen_droplogin_subcmds(const char *login)
 {

contrib/babelfishpg_tsql/src/rolecmds.h

@@ -81,5 +81,7 @@ extern char *convertToUPN(char *input);
 extern bool windows_login_contains_invalid_chars(char *input);
 extern bool windows_domain_contains_invalid_chars(char *input);
 extern bool check_windows_logon_length(char *input);
+extern void grant_dbo_to_login(const char* login, const char* db_name);
+extern void revoke_dbo_from_login(const char* login, const char* db_name);
 
 #endif