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

…lfish-for-postgresql#2925)

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 arised 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.

Task: BABEL-5119, BABEL-5218

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

contrib/babelfishpg_tsql/runtime/functions.c

@@ -58,6 +58,7 @@
 #include "../src/catalog.h"
 #include "../src/timezone.h"
 #include "../src/collation.h"
+#include "../src/dbcmds.h"
 #include "../src/rolecmds.h"
 #include "utils/fmgroids.h"
 #include "utils/acl.h"
@@ -2620,6 +2621,7 @@ has_dbaccess(PG_FUNCTION_ARGS)
 	const char *user = NULL;
 	const char *login;
 	int16		db_id;
+	bool		login_is_db_owner;
 
 	i = strlen(lowercase_db_name);
 	while (i > 0 && isspace((unsigned char) lowercase_db_name[i - 1]))
@@ -2632,6 +2634,7 @@ has_dbaccess(PG_FUNCTION_ARGS)
 
 	login = GetUserNameFromId(GetSessionUserId(), false);
 	user = get_authid_user_ext_physical_name(lowercase_db_name, login);
+	login_is_db_owner = 0 == strncmp(login, get_owner_of_db(lowercase_db_name), NAMEDATALEN);
 
 	/*
 	 * Special cases: Database Owner should always have access If this DB has
@@ -2642,7 +2645,11 @@ has_dbaccess(PG_FUNCTION_ARGS)
 		Oid			datdba;
 
 		datdba = get_role_oid("sysadmin", false);
-		if (is_member_of_role(GetSessionUserId(), datdba))
+		if (is_member_of_role(GetSessionUserId(), datdba) || login_is_db_owner)
+			/* 
+			 * The login will have access to the database if it is a member
+			 * of sysadmin or it is the owner of the database.
+			 */
 			user = get_dbo_role_name(lowercase_db_name);
 		else
 		{

contrib/babelfishpg_tsql/sql/upgrades/babelfishpg_tsql--3.7.0--3.8.0.sql

@@ -11,6 +11,36 @@ SELECT set_config('search_path', 'sys, '||current_setting('search_path'), false)
  * final behaviour.
  */
 
+-- Assigning dbo role to the db_owner login
+DO $$
+DECLARE
+    owner_name NAME;
+    db_name TEXT;
+    role_name NAME;
+    owner_cursor CURSOR FOR SELECT DISTINCT owner, name FROM sys.babelfish_sysdatabases;
+BEGIN
+    OPEN owner_cursor;
+    FETCH NEXT FROM owner_cursor INTO owner_name, db_name;
+
+    WHILE FOUND
+    LOOP
+        SELECT rolname FROM sys.babelfish_authid_user_ext WHERE database_name = db_name INTO role_name;
+
+        IF db_name = 'master' OR db_name = 'tempdb' OR db_name = 'msdb'
+        THEN
+            FETCH NEXT FROM owner_cursor INTO owner_name, db_name;
+            CONTINUE;
+        END IF;
+
+        EXECUTE FORMAT('GRANT %I TO %I', role_name, owner_name);
+
+        FETCH NEXT FROM owner_cursor INTO owner_name, db_name;
+    END LOOP;
+
+    CLOSE owner_cursor;
+END;
+$$ LANGUAGE plpgsql;
+
 CREATE OR REPLACE FUNCTION bbf_string_agg_finalfn_varchar(INTERNAL)
 RETURNS sys.VARCHAR
 AS 'string_agg_finalfn' LANGUAGE INTERNAL;

contrib/babelfishpg_tsql/src/dbcmds.c

@@ -50,7 +50,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,
@@ -98,14 +99,17 @@ have_createdb_privilege(void)
  * 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
@@ -116,6 +120,7 @@ gen_createdb_subcmds(const char *schema, const char *dbo, const char *db_owner,
 	appendStringInfo(&query, "CREATE ROLE dummy 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)
 	{
@@ -136,9 +141,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,
@@ -156,6 +161,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++);
@@ -524,7 +542,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);
 
 	/* Set current user to session user for create permissions */
 	prev_current_user = GetUserNameFromId(GetUserId(), false);

contrib/babelfishpg_tsql/src/pl_exec-2.c

@@ -20,6 +20,7 @@
 #include "catalog.h"
 #include "dbcmds.h"
 #include "pl_explain.h"
+#include "rolecmds.h"
 #include "session.h"
 
 /* helper function to get current T-SQL estate */
@@ -3869,9 +3870,14 @@ 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);	
+
+	/* Revoke dbo role from the previous owner */
+	grant_revoke_dbo_to_login(get_owner_of_db(stmt->db_name), stmt->db_name, false);
+
+	/* Grant dbo role to the new owner */
+	grant_revoke_dbo_to_login(stmt->new_owner_name, stmt->db_name, true);
+	update_db_owner(stmt->new_owner_name, stmt->db_name);
+
 	return PLTSQL_RC_OK;
 }
 

contrib/babelfishpg_tsql/src/pl_exec.c

@@ -4661,7 +4661,10 @@ exec_stmt_execsql(PLtsql_execstate *estate,
 		 * schemas, Change the session property.
 		 */
 		if (stmt->schema_name != NULL && (strcmp(stmt->schema_name, "sys") == 0 || strcmp(stmt->schema_name, "information_schema") == 0))
+		{
 			set_session_properties(stmt->db_name);
+			SetCurrentRoleId(GetSessionUserId(), false);
+		}
 	}
 	if (stmt->is_dml || stmt->is_ddl || stmt->is_create_view)
 	{

contrib/babelfishpg_tsql/src/rolecmds.c

@@ -581,6 +581,83 @@ grant_guests_to_login(const char *login)
 	pfree(query.data);
 }
 
+/* 
+ * Grant/revoke dbo role from the login.
+ * The 'is_grant' flag determines if the action is grant/revoke.
+ */
+void
+grant_revoke_dbo_to_login(const char* login, const char* db_name, bool is_grant)
+{
+	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);
+
+	if (is_grant)
+	{
+		/* Build dummy GRANT statement to grant membership to login  */
+		appendStringInfo(&query, "GRANT dummy TO dummy; ");
+	}
+	else
+	{
+		/* 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_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 = 23;
+
+	/* do this step */
+	ProcessUtility(wrapper,
+				   "(ALTER DATABASE OWNER )",
+				   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

@@ -82,6 +82,6 @@ extern bool windows_domain_contains_invalid_chars(char *input);
 extern bool check_windows_logon_length(char *input);
 extern char* get_windows_domain_name(char* input);
 extern bool windows_domain_is_not_supported(char* domain_name);
-
+extern void grant_revoke_dbo_to_login(const char* login, const char* db_name, bool is_grant);
 
 #endif

test/JDBC/expected/BABEL-5119-vu-cleanup.out

@@ -0,0 +1,27 @@
+-- tsql
+USE master
+GO
+DROP TABLE guest.BABEL5119_t4
+GO
+DROP TABLE guest.BABEL5119_t5
+GO
+DROP TABLE guest.BABEL5119_t6
+GO
+DROP VIEW guest.BABEL5119_v4
+GO
+DROP VIEW guest.BABEL5119_v5
+GO
+DROP VIEW guest.BABEL5119_v6
+GO
+DROP PROCEDURE guest.BABEL5119_p4
+GO
+DROP PROCEDURE guest.BABEL5119_p5
+GO
+DROP PROCEDURE guest.BABEL5119_p6
+GO
+DROP LOGIN login_babel5119_1
+GO
+DROP LOGIN login_babel5119_2
+GO
+DROP DATABASE BABEL5119_db
+GO

test/JDBC/expected/BABEL-5119-vu-prepare.out

@@ -0,0 +1,25 @@
+-- tsql
+CREATE DATABASE BABEL5119_db
+GO
+
+USE BABEL5119_db
+GO
+CREATE TABLE BABEL5119_t1(a int)
+GO
+CREATE TABLE BABEL5119_t2(a int)
+GO
+CREATE TABLE BABEL5119_t3(a int)
+GO
+CREATE VIEW BABEL5119_v1 AS SELECT 1
+GO
+CREATE VIEW BABEL5119_v2 AS SELECT 1
+GO
+CREATE VIEW BABEL5119_v3 AS SELECT 1
+GO
+CREATE PROCEDURE BABEL5119_p1 AS SELECT 1
+GO
+CREATE PROCEDURE BABEL5119_p2 AS SELECT 1
+GO
+CREATE PROCEDURE BABEL5119_p3 AS SELECT 1
+GO
+-- terminate-tsql-conn