Support fixed database role db_ddladmin (#101) (#476)

Description Engine changes for babelfish fixed database role db_ddladmin. Implement hooks to allow members of db_ddladmin to ALTER/DROP on database objects. Handle dump restore changes. Extensions PR : amazon-aurora/babelfish_extensions#71 Issues Resolved: BABEL-5116 Signed-off-by: Tanzeel Khan <[email protected]>
babelfish-for-postgresql · Nov 19, 2024 · 7e264f2 · 7e264f2
1 parent 2e0148e
commit 7e264f2
Show file tree

Hide file tree

Showing 12 changed files with 65 additions and 23 deletions.
diff --git a/src/backend/catalog/pg_proc.c b/src/backend/catalog/pg_proc.c
@@ -123,6 +123,11 @@ ProcedureCreate(const char *procedureName,
 	 */
 	Assert(PointerIsValid(prosrc));
 
+	if (pltsql_get_object_owner_hook)
+	{
+		proowner = (*pltsql_get_object_owner_hook)(procNamespace, proowner);
+	}
+
 	parameterCount = parameterTypes->dim1;
 	if (parameterCount < 0 || parameterCount > FUNC_MAX_ARGS)
 		ereport(ERROR,

diff --git a/src/backend/catalog/pg_type.c b/src/backend/catalog/pg_type.c
@@ -253,6 +253,11 @@ TypeCreate(Oid newTypeOid,
 				 errmsg("invalid type internal size %d",
 						internalSize)));
 
+	if (pltsql_get_object_owner_hook)
+	{
+		ownerId = (*pltsql_get_object_owner_hook)(typeNamespace, ownerId);
+	}
+
 	if (passedByValue)
 	{
 		/*

diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c
@@ -219,7 +219,8 @@ AlterObjectRename_internal(Relation rel, Oid objectId, const char *new_name)
 		Assert(!isnull);
 		ownerId = DatumGetObjectId(datum);
 
-		if (!has_privs_of_role(GetUserId(), DatumGetObjectId(ownerId)))
+		if (!has_privs_of_role(GetUserId(), DatumGetObjectId(ownerId)) &&
+			!(OidIsValid(namespaceId) && IS_BBF_DB_DDLADMIN(namespaceId)))
 			aclcheck_error(ACLCHECK_NOT_OWNER, get_object_type(classId, objectId),
 						   old_name);
 

diff --git a/src/backend/commands/dropcmds.c b/src/backend/commands/dropcmds.c
@@ -104,8 +104,11 @@ RemoveObjects(DropStmt *stmt)
 
 		/* Check permissions. */
 		namespaceId = get_object_namespace(&address);
-		if (!OidIsValid(namespaceId) ||
-			!object_ownercheck(NamespaceRelationId, namespaceId, GetUserId()))
+		if ((!OidIsValid(namespaceId) ||
+			!object_ownercheck(NamespaceRelationId, namespaceId, GetUserId())) &&
+			!(OidIsValid(namespaceId) && IS_BBF_DB_DDLADMIN(namespaceId)) &&
+			!(stmt->removeType == OBJECT_SCHEMA && IS_BBF_DB_DDLADMIN(address.objectId)) &&
+			!(relation != NULL && IS_BBF_DB_DDLADMIN(RelationGetNamespace(relation))))
 			check_object_ownership(GetUserId(), stmt->removeType, address,
 								   object, relation);
 

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
@@ -847,6 +847,12 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 				 errmsg("only shared relations can be placed in pg_global tablespace")));
 
+
+	if (pltsql_get_object_owner_hook && !OidIsValid(ownerId))
+	{
+		ownerId = (*pltsql_get_object_owner_hook)(namespaceId, ownerId);
+	}
+
 	/* Identify user ID that will own the table */
 	if (!OidIsValid(ownerId))
 		ownerId = GetUserId();
@@ -1743,7 +1749,8 @@ RangeVarCallbackForDropRelation(const RangeVar *rel, Oid relOid, Oid oldRelOid,
 
 	/* Allow DROP to either table owner or schema owner */
 	if (!object_ownercheck(RelationRelationId, relOid, GetUserId()) &&
-		!object_ownercheck(NamespaceRelationId, classform->relnamespace, GetUserId()))
+		!object_ownercheck(NamespaceRelationId, classform->relnamespace, GetUserId()) &&
+		!IS_BBF_DB_DDLADMIN(classform->relnamespace))
 		aclcheck_error(ACLCHECK_NOT_OWNER,
 					   get_relkind_objtype(classform->relkind),
 					   rel->relname);
@@ -2697,7 +2704,8 @@ MergeAttributes(List *columns, const List *supers, char relpersistence,
 		 * We should have an UNDER permission flag for this, but for now,
 		 * demand that creator of a child table own the parent.
 		 */
-		if (!object_ownercheck(RelationRelationId, RelationGetRelid(relation), GetUserId()))
+		if (!object_ownercheck(RelationRelationId, RelationGetRelid(relation), GetUserId()) &&
+			!IS_BBF_DB_DDLADMIN(RelationGetNamespace(relation)))
 			aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(relation->rd_rel->relkind),
 						   RelationGetRelationName(relation));
 
@@ -3743,7 +3751,8 @@ renameatt_check(Oid myrelid, Form_pg_class classform, bool recursing)
 	/*
 	 * permissions checking.  only the owner of a class can change its schema.
 	 */
-	if (!object_ownercheck(RelationRelationId, myrelid, GetUserId()))
+	if (!object_ownercheck(RelationRelationId, myrelid, GetUserId()) &&
+		!IS_BBF_DB_DDLADMIN(get_rel_namespace(myrelid)))
 		aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(myrelid)),
 					   NameStr(classform->relname));
 	if (!allowSystemTableMods && IsSystemClass(myrelid, classform))
@@ -6680,7 +6689,8 @@ ATSimplePermissions(AlterTableType cmdtype, Relation rel, int allowed_targets)
 	}
 
 	/* Permissions checks */
-	if (!object_ownercheck(RelationRelationId, RelationGetRelid(rel), GetUserId()))
+	if (!object_ownercheck(RelationRelationId, RelationGetRelid(rel), GetUserId()) &&
+		!IS_BBF_DB_DDLADMIN(RelationGetNamespace(rel)))
 		aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->rd_rel->relkind),
 					   RelationGetRelationName(rel));
 
@@ -12178,7 +12188,7 @@ checkFkeyPermissions(Relation rel, int16 *attnums, int natts)
 	/* Okay if we have relation-level REFERENCES permission */
 	aclresult = pg_class_aclcheck(RelationGetRelid(rel), roleid,
 								  ACL_REFERENCES);
-	if (aclresult == ACLCHECK_OK)
+	if (aclresult == ACLCHECK_OK || IS_BBF_DB_DDLADMIN(RelationGetNamespace(rel)))
 		return;
 	/* Else we must have REFERENCES on each column */
 	for (i = 0; i < natts; i++)
@@ -17756,7 +17766,8 @@ RangeVarCallbackOwnsRelation(const RangeVar *relation,
 	if (!HeapTupleIsValid(tuple))	/* should not happen */
 		elog(ERROR, "cache lookup failed for relation %u", relId);
 
-	if (!object_ownercheck(RelationRelationId, relId, GetUserId()))
+	if (!object_ownercheck(RelationRelationId, relId, GetUserId()) &&
+		!IS_BBF_DB_DDLADMIN(get_rel_namespace(relId)))
 		aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(relId)),
 					   relation->relname);
 
@@ -17792,7 +17803,8 @@ RangeVarCallbackForAlterRelation(const RangeVar *rv, Oid relid, Oid oldrelid,
 	relkind = classform->relkind;
 
 	/* Must own relation. */
-	if (!object_ownercheck(RelationRelationId, relid, GetUserId()))
+	if (!object_ownercheck(RelationRelationId, relid, GetUserId()) &&
+		!IS_BBF_DB_DDLADMIN(classform->relnamespace))
 		aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(relid)), rv->relname);
 
 	/* No system table modifications unless explicitly allowed. */

diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
@@ -376,7 +376,7 @@ CreateTriggerFiringOn(CreateTrigStmt *stmt, const char *queryString,
 	{
 		aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
 									  ACL_TRIGGER);
-		if (aclresult != ACLCHECK_OK)
+		if (aclresult != ACLCHECK_OK && !IS_BBF_DB_DDLADMIN(RelationGetNamespace(rel)))
 			aclcheck_error(aclresult, get_relkind_objtype(rel->rd_rel->relkind),
 						   RelationGetRelationName(rel));
 
@@ -751,7 +751,7 @@ CreateTriggerFiringOn(CreateTrigStmt *stmt, const char *queryString,
 	if (!isInternal)
 	{
 		aclresult = object_aclcheck(ProcedureRelationId, funcoid, GetUserId(), ACL_EXECUTE);
-		if (aclresult != ACLCHECK_OK)
+		if (aclresult != ACLCHECK_OK && !IS_BBF_DB_DDLADMIN(get_func_namespace(funcoid)))
 			aclcheck_error(aclresult, OBJECT_FUNCTION,
 						   NameListToString(stmt->funcname));
 	}
@@ -1569,7 +1569,8 @@ RangeVarCallbackForRenameTrigger(const RangeVar *rv, Oid relid, Oid oldrelid,
 				 errdetail_relkind_not_supported(form->relkind)));
 
 	/* you must own the table to rename one of its triggers */
-	if (!object_ownercheck(RelationRelationId, relid, GetUserId()))
+	if (!object_ownercheck(RelationRelationId, relid, GetUserId()) &&
+		!IS_BBF_DB_DDLADMIN(get_rel_namespace(relid)))
 		aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(relid)), rv->relname);
 	if (!allowSystemTableMods && IsSystemClass(relid, form))
 		ereport(ERROR,

diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
@@ -3787,7 +3787,8 @@ RenameType(RenameStmt *stmt)
 	typTup = (Form_pg_type) GETSTRUCT(tup);
 
 	/* check permissions on type */
-	if (!object_ownercheck(TypeRelationId, typeOid, GetUserId()))
+	if (!object_ownercheck(TypeRelationId, typeOid, GetUserId()) &&
+		!IS_BBF_DB_DDLADMIN(typTup->typnamespace))
 		aclcheck_error_type(ACLCHECK_NOT_OWNER, typeOid);
 
 	/* ALTER DOMAIN used on a non-domain? */

diff --git a/src/backend/rewrite/rewriteDefine.c b/src/backend/rewrite/rewriteDefine.c
@@ -271,7 +271,8 @@ DefineQueryRewrite(const char *rulename,
 	/*
 	 * Check user has permission to apply rules to this relation.
 	 */
-	if (!object_ownercheck(RelationRelationId, event_relid, GetUserId()))
+	if (!object_ownercheck(RelationRelationId, event_relid, GetUserId()) &&
+		!IS_BBF_DB_DDLADMIN(RelationGetNamespace(event_relation)))
 		aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(event_relation->rd_rel->relkind),
 					   RelationGetRelationName(event_relation));
 

diff --git a/src/backend/utils/adt/acl.c b/src/backend/utils/adt/acl.c
@@ -132,6 +132,8 @@ static void RoleMembershipCacheCallback(Datum arg, int cacheid, uint32 hashvalue
 
 bbf_get_sysadmin_oid_hook_type bbf_get_sysadmin_oid_hook = NULL;
 get_bbf_admin_oid_hook_type get_bbf_admin_oid_hook = NULL;
+pltsql_get_object_owner_hook_type pltsql_get_object_owner_hook = NULL;
+is_bbf_db_ddladmin_operation_hook_type is_bbf_db_ddladmin_operation_hook = NULL;
 
 
 /*

diff --git a/src/bin/pg_dump/dump_babel_utils.c b/src/bin/pg_dump/dump_babel_utils.c
@@ -55,13 +55,13 @@ static babelfish_status bbf_status = NONE;
 static char *default_bbf_db_principals =
 			"('master_dbo', 'master_db_owner', 'master_guest', "
 			"'master_db_accessadmin', 'master_db_securityadmin', "
-			"'master_db_datareader', 'master_db_datawriter', "
+			"'master_db_datareader', 'master_db_datawriter', 'master_db_ddladmin', "
 			"'msdb_dbo', 'msdb_db_owner', 'msdb_guest', "
 			"'msdb_db_accessadmin', 'msdb_db_securityadmin', "
-			"'msdb_db_datareader', 'msdb_db_datawriter', "
+			"'msdb_db_datareader', 'msdb_db_datawriter', 'msdb_db_ddladmin', "
 			"'tempdb_dbo', 'tempdb_db_owner', 'tempdb_guest', "
 			"'tempdb_db_accessadmin', 'tempdb_db_securityadmin', "
-			"'tempdb_db_datareader', 'tempdb_db_datawriter')" ;
+			"'tempdb_db_datareader', 'tempdb_db_datawriter', 'tempdb_db_ddladmin')";
 
 
 
@@ -2002,7 +2002,7 @@ dumpBabelPhysicalDatabaseACLs(Archive *fout)
 					"\n	SET LOCAL ROLE sysadmin;"
 					"\n	FOR rolname, original_name IN ("
 					"\n		SELECT a.rolname, a.orig_username FROM sys.babelfish_authid_user_ext a"
-					"\n			WHERE orig_username IN ('dbo','db_accessadmin','db_securityadmin') AND"
+					"\n			WHERE orig_username IN ('dbo','db_accessadmin','db_securityadmin','db_ddladmin') AND"
 					"\n			database_name NOT IN ('master', 'tempdb', 'msdb')");
 
 	if (bbf_db_name)
@@ -2013,7 +2013,7 @@ dumpBabelPhysicalDatabaseACLs(Archive *fout)
 					"\n	) LOOP"
 					"\n		CASE WHEN original_name = 'dbo' THEN"
 					"\n			EXECUTE format('GRANT CREATE, CONNECT, TEMPORARY ON DATABASE \"%%s\" TO \"%%s\"; ', CURRENT_DATABASE(), rolname);"
-					"\n		WHEN original_name IN ('db_securityadmin','db_accessadmin') THEN"
+					"\n		WHEN original_name IN ('db_securityadmin','db_accessadmin','db_ddladmin') THEN"
 					"\n			EXECUTE format('GRANT CREATE ON DATABASE \"%%s\" TO \"%%s\"; ', CURRENT_DATABASE(), rolname);"
 					"\n		END CASE;"
 					"\n	END LOOP;"

diff --git a/src/bin/pg_dump/dumpall_babel_utils.c b/src/bin/pg_dump/dumpall_babel_utils.c
@@ -37,14 +37,14 @@ static babelfish_status bbf_status = NONE;
 
 static char default_bbf_roles[] = "('sysadmin', 'bbf_role_admin', 'securityadmin', 'dbcreator', "
                                   "'master_dbo', 'master_db_owner', 'master_guest', "
-                                  "'master_db_accessadmin', 'master_db_securityadmin', "
+                                  "'master_db_accessadmin', 'master_db_securityadmin', 'master_db_ddladmin', "
                                   "'master_db_datareader', 'master_db_datawriter', "
                                   "'msdb_dbo', 'msdb_db_owner', 'msdb_guest', "
                                   "'msdb_db_accessadmin', 'msdb_db_securityadmin', "
-                                  "'msdb_db_datareader', 'msdb_db_datawriter', "
+                                  "'msdb_db_datareader', 'msdb_db_datawriter', 'msdb_db_ddladmin', "
                                   "'tempdb_dbo', 'tempdb_db_owner', 'tempdb_guest', "
                                   "'tempdb_db_accessadmin', 'tempdb_db_securityadmin', "
-                                  "'tempdb_db_datareader', 'tempdb_db_datawriter')" ;
+                                  "'tempdb_db_datareader', 'tempdb_db_datawriter', 'tempdb_db_ddladmin')";
 
 /*
  * Run a query, return the results, exit program on failure.

diff --git a/src/include/utils/acl.h b/src/include/utils/acl.h
@@ -300,4 +300,15 @@ extern PGDLLEXPORT get_bbf_admin_oid_hook_type get_bbf_admin_oid_hook;
 typedef void (*bbf_execute_grantstmt_as_dbsecadmin_hook_type) (ObjectType objType, Oid objId, Oid ownerId, AclMode privileges, Oid *grantorId, AclMode *grantOptions);
 extern PGDLLEXPORT bbf_execute_grantstmt_as_dbsecadmin_hook_type bbf_execute_grantstmt_as_dbsecadmin_hook;
 
+typedef Oid (*pltsql_get_object_owner_hook_type) (Oid, Oid);
+extern PGDLLEXPORT pltsql_get_object_owner_hook_type pltsql_get_object_owner_hook;
+
+typedef bool (*is_bbf_db_ddladmin_operation_hook_type) (Oid namespaceId);
+extern PGDLLEXPORT is_bbf_db_ddladmin_operation_hook_type is_bbf_db_ddladmin_operation_hook;
+
+#define IS_BBF_DB_DDLADMIN(namespaceId) \
+	(is_bbf_db_ddladmin_operation_hook &&       \
+	 is_bbf_db_ddladmin_operation_hook(namespaceId))
+
+
 #endif							/* ACL_H */