grant on schema does not take effect on future objects created in a s…

…chema by schema's authorized users (#2993)

grant on schema does not take effect on future objects created in a schema by the schema's authorized users

When a user u1 is granted ownership on a schema s1 at the time of creation. And there is grant select on schema permission on s1 to u2. The future objects created by the users dbo and u1 should be accessible by the user u2

Earlier, if u1 created a table in s1 schema since it's authorized to do so, the user u2 was not allowed to access the object created by u1.

With this commit, future objects created by u1 and dbo will be accessible to the user u2.

Task: BABEL-5172

Signed-off-by: Shalini Lohia [email protected]

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

@@ -10489,6 +10489,17 @@ BEGIN
 END;
 $$ LANGUAGE plpgsql STABLE PARALLEL SAFE;
 
+-- This is a temporary procedure which is called during upgrade to alter
+-- default privileges on all the schemas where the schema owner is not dbo/db_owner
+CREATE OR REPLACE PROCEDURE sys.babelfish_alter_default_privilege_on_schema()
+LANGUAGE C
+AS 'babelfishpg_tsql', 'alter_default_privilege_on_schema';
+
+CALL sys.babelfish_alter_default_privilege_on_schema();
+
+-- Drop this procedure after it gets executed once.
+DROP PROCEDURE sys.babelfish_alter_default_privilege_on_schema();
+
 -- Drops the temporary procedure used by the upgrade script.
 -- Please have this be one of the last statements executed in this upgrade script.
 DROP PROCEDURE sys.babelfish_drop_deprecated_object(varchar, varchar, varchar);

contrib/babelfishpg_tsql/src/catalog.c

@@ -5933,4 +5933,116 @@ clean_up_bbf_partition_metadata(int16 dbid)
 
 	systable_endscan(scan);
 	table_close(rel, RowExclusiveLock);
-}
+}
+/* 
+ * This is a temporary procedure which is called during upgrade to alter
+ * default privileges on all the schemas where the schema owner is not dbo/db_owner.
+ */
+static void
+alter_default_privilege_for_db(char *dbname)
+{
+	SysScanDesc scan;
+	Relation	bbf_schema_rel;
+	TupleDesc	dsc;
+	HeapTuple	tuple_bbf_schema;
+	ScanKeyData scanKey[2];
+	int16		dbid = get_db_id(dbname);
+	MigrationMode baseline_mode = is_user_database_singledb(dbname) ? SINGLE_DB : MULTI_DB;
+
+	/* Fetch the relation */
+	bbf_schema_rel = table_open(get_bbf_schema_perms_oid(),
+									AccessShareLock);
+	dsc = RelationGetDescr(bbf_schema_rel);
+	ScanKeyInit(&scanKey[0],
+				Anum_bbf_schema_perms_dbid,
+				BTEqualStrategyNumber, F_INT2EQ,
+				Int16GetDatum(dbid));
+	ScanKeyEntryInitialize(&scanKey[1], 0,
+				Anum_bbf_schema_perms_object_type,
+				BTEqualStrategyNumber,
+				InvalidOid,
+				tsql_get_database_or_server_collation_oid_internal(false),
+				F_TEXTEQ,
+				CStringGetTextDatum(OBJ_SCHEMA));
+
+	scan = systable_beginscan(bbf_schema_rel, get_bbf_schema_perms_idx_oid(),
+							true, NULL, 2, scanKey);
+	tuple_bbf_schema = systable_getnext(scan);
+
+	while (HeapTupleIsValid(tuple_bbf_schema))
+	{
+		bool		isnull;
+		const char	*schema_name;
+		const char	*grantee;
+		int			current_permission;
+		char		*schema_owner;
+		char		*physical_schema;
+		const char	*dbo_user;
+		const char	*db_owner;
+		int			i;
+
+		schema_name = TextDatumGetCString(heap_getattr(tuple_bbf_schema, Anum_bbf_schema_perms_schema_name, dsc, &isnull));
+		grantee = TextDatumGetCString(heap_getattr(tuple_bbf_schema, Anum_bbf_schema_perms_grantee, dsc, &isnull));
+		current_permission = DatumGetInt32(heap_getattr(tuple_bbf_schema, Anum_bbf_schema_perms_permission, dsc, &isnull));
+
+		physical_schema = get_physical_schema_name_by_mode(dbname, schema_name, baseline_mode);
+		dbo_user = get_dbo_role_name_by_mode(dbname, baseline_mode);
+		db_owner = get_db_owner_name_by_mode(dbname, baseline_mode);
+		schema_owner = GetUserNameFromId(get_owner_of_schema(physical_schema), false);
+
+		/* If schema owner is other that dbo or db_owner user, only then execute ALTER DEFAULT PRIVILEGES. */
+		if ((strcmp(schema_owner, dbo_user) != 0) && (strcmp(schema_owner, db_owner) != 0))
+		{
+			/* For each permission, grant alter default privileges explicitly. */
+			for (i = 0; i < NUMBER_OF_PERMISSIONS; i++)
+			{
+				if ((current_permission & permissions[i]) &&  permissions[i] != ACL_EXECUTE)
+				{
+					char	*alter_query = NULL;
+					char	*grant_query = NULL;
+					alter_query = psprintf("ALTER DEFAULT PRIVILEGES FOR ROLE %s, %s IN SCHEMA %s GRANT %s ON TABLES TO %s", dbo_user, schema_owner, physical_schema, privilege_to_string(permissions[i]), grantee);
+					exec_utility_cmd_helper(alter_query);
+					grant_query = psprintf("GRANT %s ON ALL TABLES IN SCHEMA %s TO %s", privilege_to_string(permissions[i]), physical_schema, grantee);
+					exec_utility_cmd_helper(grant_query);
+					pfree(alter_query);
+					pfree(grant_query);
+				}
+			}
+		}
+		pfree(physical_schema);
+		pfree(schema_owner);
+		tuple_bbf_schema = systable_getnext(scan);
+	}
+
+	systable_endscan(scan);
+	table_close(bbf_schema_rel, AccessShareLock);
+}
+
+
+PG_FUNCTION_INFO_V1(alter_default_privilege_on_schema);
+Datum
+alter_default_privilege_on_schema(PG_FUNCTION_ARGS)
+{
+	Relation	db_rel;
+	TableScanDesc scan;
+	HeapTuple	tuple;
+	bool		is_null;
+
+	db_rel = table_open(sysdatabases_oid, AccessShareLock);
+	scan = table_beginscan_catalog(db_rel, 0, NULL);
+	tuple = heap_getnext(scan, ForwardScanDirection);
+
+	while (HeapTupleIsValid(tuple))
+	{
+		Datum		db_name_datum = heap_getattr(tuple, Anum_sysdatabases_name,
+												 db_rel->rd_att, &is_null);
+		char *db_name = TextDatumGetCString(db_name_datum);
+
+		alter_default_privilege_for_db(db_name);
+		pfree(db_name);
+		tuple = heap_getnext(scan, ForwardScanDirection);
+	}
+	table_endscan(scan);
+	table_close(db_rel, AccessShareLock);
+	PG_RETURN_INT32(0);
+}

contrib/babelfishpg_tsql/src/multidb.c

@@ -1368,13 +1368,13 @@ get_dbo_schema_name(const char *dbname)
 }
 
 char *
-get_dbo_role_name(const char *dbname)
+get_dbo_role_name_by_mode(const char *dbname, MigrationMode mode)
 {
 	char	   *name = palloc0(MAX_BBF_NAMEDATALEND);
 
 	Assert(dbname != NULL);
 
-	if (SINGLE_DB == get_migration_mode() && 0 != strcmp(dbname, "master") 
+	if (SINGLE_DB == mode && 0 != strcmp(dbname, "master") 
 	                    && 0 != strcmp(dbname, "tempdb") && 0 != strcmp(dbname, "msdb"))
 	{	
 		snprintf(name, MAX_BBF_NAMEDATALEND, "%s", "dbo");
@@ -1388,13 +1388,19 @@ get_dbo_role_name(const char *dbname)
 }
 
 char *
-get_db_owner_name(const char *dbname)
+get_dbo_role_name(const char *dbname)
+{
+	return get_dbo_role_name_by_mode(dbname, get_migration_mode());
+}
+
+char *
+get_db_owner_name_by_mode(const char *dbname, MigrationMode	mode)
 {
 	char	   *name = palloc0(MAX_BBF_NAMEDATALEND);
 
 	Assert(dbname != NULL);
 
-	if (SINGLE_DB == get_migration_mode() && 0 != strcmp(dbname, "master") 
+	if (SINGLE_DB == mode && 0 != strcmp(dbname, "master") 
 	                    && 0 != strcmp(dbname, "tempdb") && 0 != strcmp(dbname, "msdb"))
 	{	
 		snprintf(name, MAX_BBF_NAMEDATALEND, "%s", "db_owner");
@@ -1407,6 +1413,12 @@ get_db_owner_name(const char *dbname)
 	return name;
 }
 
+char *
+get_db_owner_name(const char *dbname)
+{
+	return get_db_owner_name_by_mode(dbname, get_migration_mode());
+}
+
 char *
 get_guest_role_name(const char *dbname)
 {

contrib/babelfishpg_tsql/src/multidb.h

@@ -22,7 +22,9 @@ extern char *get_physical_schema_name(char *db_name, const char *schema_name);
 extern char *get_physical_schema_name_by_mode(char *db_name, const char *schema_name, MigrationMode mode);
 extern char *get_dbo_schema_name(const char *dbname);
 extern char *get_dbo_role_name(const char *dbname);
+extern char *get_dbo_role_name_by_mode(const char *dbname, MigrationMode mode);
 extern char *get_db_owner_name(const char *dbname);
+extern char *get_db_owner_name_by_mode(const char *dbname, MigrationMode mode);
 extern char *get_guest_role_name(const char *dbname);
 extern char *get_guest_schema_name(const char *dbname);
 extern bool is_shared_schema(const char *name);

contrib/babelfishpg_tsql/src/pl_exec-2.c

@@ -3804,7 +3804,7 @@ exec_stmt_grantschema(PLtsql_execstate *estate, PLtsql_stmt_grantschema *stmt)
 		for (i = 0; i < NUMBER_OF_PERMISSIONS; i++)
 		{
 			if (stmt->privileges & permissions[i])
-				exec_grantschema_subcmds(schema_name, rolname, stmt->is_grant, stmt->with_grant_option, permissions[i]);
+				exec_grantschema_subcmds(schema_name, rolname, stmt->is_grant, stmt->with_grant_option, permissions[i], false);
 		}
 
 		if (stmt->is_grant)

contrib/babelfishpg_tsql/src/pl_handler.c

@@ -3614,7 +3614,7 @@ bbf_ProcessUtility(PlannedStmt *pstmt,
 						for (i = 0; i < NUMBER_OF_PERMISSIONS; i++)
 						{
 							/* Execute the GRANT SCHEMA subcommands. */
-							exec_grantschema_subcmds(create_schema->schemaname, rolspec->rolename, true, false, permissions[i]);
+							exec_grantschema_subcmds(create_schema->schemaname, rolspec->rolename, true, false, permissions[i], true);
 						}
 					}
 					return;

contrib/babelfishpg_tsql/src/pltsql.h

@@ -2062,6 +2062,7 @@ extern int	get_insert_bulk_kilobytes_per_batch(void);
 extern char *get_original_query_string(void);
 extern AclMode string_to_privilege(const char *privname);
 extern const char *privilege_to_string(AclMode privilege);
+extern Oid get_owner_of_schema(const char *schema);
 
 /*
  * Functions for namespace handling in pl_funcs.c
@@ -2128,7 +2129,7 @@ extern const char *gen_schema_name_for_fulltext_index(const char *schema_name);
 extern bool check_fulltext_exist(const char *schema_name, const char *table_name);
 extern char *replace_special_chars_fts_impl(char *input_str);
 extern bool is_unique_index(Oid relid, const char *index_name);
-extern void exec_grantschema_subcmds(const char *schema, const char *rolname, bool is_grant, bool with_grant_option, AclMode privilege);
+extern void exec_grantschema_subcmds(const char *schema, const char *rolname, bool is_grant, bool with_grant_option, AclMode privilege, bool is_create_schema);
 extern int	TsqlUTF8LengthInUTF16(const void *vin, int len);
 extern void TsqlCheckUTF16Length_bpchar(const char *s, int32 len, int32 maxlen, int charlen, bool isExplicit);
 extern void TsqlCheckUTF16Length_varchar(const char *s, int32 len, int32 maxlen, bool isExplicit);
@@ -2171,6 +2172,7 @@ extern void update_GrantRoleStmt(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);
+extern void update_AlterDefaultPrivilegesStmt(Node *n, const char *schema, const char *role1, const char *role2, const char *grantee, const char *priv);
 extern AccessPriv *make_accesspriv_node(const char *priv_name);
 extern RoleSpec   *make_rolespec_node(const char *rolename);
 extern void pltsql_check_or_set_default_typmod(TypeName *typeName, int32 *typmod, bool is_cast);