pg_upgrade fails with non-standard ACL

src/bin/pg_upgrade/check.c

@@ -16,6 +16,7 @@
 
 static void check_new_cluster_is_empty(void);
 static void check_databases_are_compatible(void);
+static void check_for_changed_signatures(void);
 static void check_locale_and_encoding(DbInfo *olddb, DbInfo *newdb);
 static bool equivalent_locale(int category, const char *loca, const char *locb);
 static void check_is_install_user(ClusterInfo *cluster);
@@ -142,6 +143,8 @@ check_and_dump_old_cluster(bool live_check)
 	if (GET_MAJOR_VERSION(old_cluster.major_version) <= 804)
 		new_9_0_populate_pg_largeobject_metadata(&old_cluster, true);
 
+	get_non_default_acl_infos(&old_cluster);
+
 	/*
 	 * While not a check option, we do this now because this is the only time
 	 * the old server is running.
@@ -161,6 +164,7 @@ check_new_cluster(void)
 
 	check_new_cluster_is_empty();
 	check_databases_are_compatible();
+	check_for_changed_signatures();
 
 	check_loadable_libraries();
 
@@ -443,6 +447,223 @@ check_databases_are_compatible(void)
 	}
 }
 
+/*
+ * Find the location of the last dot, return NULL if not found.
+ */
+static char *
+last_dot_location(const char *identity)
+{
+	const char *p,
+			   *ret = NULL;
+
+	for (p = identity; *p; p++)
+		if (*p == '.')
+			ret = p;
+	return unconstify(char *, ret);
+}
+
+/*
+ * check_for_changed_signatures()
+ *
+ * Check that the old cluster doesn't have non-default ACL's for system objects
+ * (relations, attributes, functions and procedures) which have different
+ * signatures in the new cluster. Otherwise generate revoke_objects.sql.
+ */
+static void
+check_for_changed_signatures(void)
+{
+	PGconn	   *conn;
+	char		subquery[QUERY_ALLOC];
+	PGresult   *res;
+	int			ntups;
+	int			i_obj_ident;
+	int			dbnum;
+	bool		need_check = false;
+	FILE	   *script = NULL;
+	bool		found_changed = false;
+	char		output_path[MAXPGPATH];
+
+	prep_status("Checking for system objects with non-default ACL");
+
+	for (dbnum = 0; dbnum < old_cluster.dbarr.ndbs; dbnum++)
+		if (old_cluster.dbarr.dbs[dbnum].non_def_acl_arr.nacls > 0)
+		{
+			need_check = true;
+			break;
+		}
+	/*
+	 * The old cluster doesn't have system objects with non-default ACL so
+	 * quickly exit.
+	 */
+	if (!need_check)
+	{
+		check_ok();
+		return;
+	}
+
+	snprintf(output_path, sizeof(output_path), "revoke_objects.sql");
+
+	snprintf(subquery, sizeof(subquery),
+		/* Get system relations which created in pg_catalog */
+		"SELECT 'pg_class'::regclass classid, oid objid, 0 objsubid "
+		"FROM pg_catalog.pg_class "
+		"WHERE relnamespace = 'pg_catalog'::regnamespace "
+		"UNION ALL "
+		/* Get system relations attributes which created in pg_catalog */
+		"SELECT 'pg_class'::regclass, att.attrelid, att.attnum "
+		"FROM pg_catalog.pg_class rel "
+		"INNER JOIN pg_catalog.pg_attribute att ON  rel.oid = att.attrelid "
+		"WHERE rel.relnamespace = 'pg_catalog'::regnamespace "
+		"UNION ALL "
+		/* Get system functions and procedure which created in pg_catalog */
+		"SELECT 'pg_proc'::regclass, oid, 0 "
+		"FROM pg_catalog.pg_proc "
+		"WHERE pronamespace = 'pg_catalog'::regnamespace ");
+
+	conn = connectToServer(&new_cluster, "template1");
+	res = executeQueryOrDie(conn,
+		"SELECT ident.type, ident.identity "
+		"FROM (%s) obj, "
+		"LATERAL pg_catalog.pg_identify_object("
+		"    obj.classid, obj.objid, obj.objsubid) ident "
+		/*
+		 * Don't rely on database collation, since we use strcmp
+		 * comparison to find non-default ACLs.
+		 */
+		"ORDER BY ident.identity COLLATE \"C\";", subquery);
+	ntups = PQntuples(res);
+
+	i_obj_ident = PQfnumber(res, "identity");
+
+	for (dbnum = 0; dbnum < old_cluster.dbarr.ndbs; dbnum++)
+	{
+		DbInfo	   *dbinfo = &old_cluster.dbarr.dbs[dbnum];
+		bool		db_used = false;
+		int			aclnum = 0,
+					objnum = 0;
+
+		/*
+		 * For every database check system objects with non-default ACL.
+		 *
+		 * AclInfo array is sorted by obj_ident. This allows us to compare
+		 * AclInfo entries with the query result above efficiently.
+		 */
+		for (aclnum = 0; aclnum < dbinfo->non_def_acl_arr.nacls; aclnum++)
+		{
+			AclInfo	   *aclinfo = &dbinfo->non_def_acl_arr.aclinfos[aclnum];
+			bool		report = false;
+
+			while (objnum < ntups)
+			{
+				int			ret;
+
+				ret = strcmp(aclinfo->obj_ident,
+							 PQgetvalue(res, objnum, i_obj_ident));
+
+				/*
+				 * The new cluster doesn't have an object with same identity,
+				 * exit the loop, report below and check next object.
+				 */
+				if (ret < 0)
+				{
+					report = true;
+					break;
+				}
+				/*
+				 * The new cluster has an object with same identity, just exit
+				 * the loop.
+				 */
+				else if (ret == 0)
+				{
+					objnum++;
+					break;
+				}
+				else
+					objnum++;
+			}
+
+			if (report)
+			{
+				found_changed = true;
+				if (script == NULL && (script = fopen_priv(output_path, "w")) == NULL)
+					pg_fatal("could not open file \"%s\": %s\n",
+							 output_path, strerror(errno));
+				if (!db_used)
+				{
+					PQExpBufferData conn_buf;
+
+					initPQExpBuffer(&conn_buf);
+					appendPsqlMetaConnect(&conn_buf, dbinfo->db_name);
+					fputs(conn_buf.data, script);
+					termPQExpBuffer(&conn_buf);
+
+					db_used = true;
+				}
+
+				/* Handle columns separately */
+				if (strstr(aclinfo->obj_type, "column") != NULL)
+				{
+					char	   *pdot = last_dot_location(aclinfo->obj_ident);
+					PQExpBufferData ident_buf;
+
+					if (pdot == NULL || *(pdot + 1) == '\0')
+						pg_fatal("invalid column identity \"%s\"",
+								 aclinfo->obj_ident);
+
+					initPQExpBuffer(&ident_buf);
+					appendBinaryPQExpBuffer(&ident_buf, aclinfo->obj_ident,
+											pdot - aclinfo->obj_ident);
+
+					fprintf(script, "REVOKE ALL (%s) ON %s FROM %s;\n",
+							/* pg_identify_object() quotes identity if necessary */
+							pdot + 1, ident_buf.data,
+							/* role_names is already quoted */
+							aclinfo->role_names);
+					termPQExpBuffer(&ident_buf);
+				}
+				/*
+				 * For relations except sequences we don't need to specify
+				 * the object type.
+				 */
+				else if (aclinfo->is_relation &&
+					strcmp(aclinfo->obj_type, "sequence") != 0)
+					fprintf(script, "REVOKE ALL ON %s FROM %s;\n",
+							/* pg_identify_object() quotes identity if necessary */
+							aclinfo->obj_ident,
+							/* role_names is already quoted */
+							aclinfo->role_names);
+				/* Other object types */
+				else
+					fprintf(script, "REVOKE ALL ON %s %s FROM %s;\n",
+							aclinfo->obj_type,
+							/* pg_identify_object() quotes identity if necessary */
+							aclinfo->obj_ident,
+							/* role_names is already quoted */
+							aclinfo->role_names);
+			}
+		}
+	}
+
+	PQclear(res);
+	PQfinish(conn);
+
+	if (script)
+		fclose(script);
+
+	if (found_changed)
+	{
+		pg_log(PG_REPORT, "fatal\n");
+		pg_fatal("Your installation contains non-default privileges for system objects\n"
+				 "for which the API has changed.  To perform the upgrade, reset these\n"
+				 "privileges to default.  The file\n"
+				 "    %s\n"
+				 "when executed by psql will revoke non-default privileges for those objects.\n\n",
+				 output_path);
+	}
+	else
+		check_ok();
+}
+
 
 /*
  * create_script_for_cluster_analyze()

src/bin/pg_upgrade/info.c

@@ -11,6 +11,7 @@
 
 #include "access/transam.h"
 #include "catalog/pg_class_d.h"
+#include "fe_utils/string_utils.h"
 #include "pg_upgrade.h"
 
 static void create_rel_filename_map(const char *old_data, const char *new_data,
@@ -23,6 +24,7 @@ static void free_db_and_rel_infos(DbInfoArr *db_arr);
 static void get_db_infos(ClusterInfo *cluster);
 static void get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo);
 static void free_rel_infos(RelInfoArr *rel_arr);
+static void free_acl_infos(AclInfoArr *acl_arr);
 static void print_db_infos(DbInfoArr *dbinfo);
 static void print_rel_infos(RelInfoArr *rel_arr);
 
@@ -328,6 +330,117 @@ get_db_and_rel_infos(ClusterInfo *cluster)
 }
 
 
+/*
+ * get_non_default_acl_infos()
+ *
+ * Fill AclInfo array with information about system objects that
+ * have non-default ACL.
+ *
+ * Note: the resulting AclInfo array is assumed to be sorted by identity.
+ */
+void
+get_non_default_acl_infos(ClusterInfo *cluster)
+{
+	int			dbnum;
+
+	for (dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)
+	{
+		DbInfo	   *dbinfo = &cluster->dbarr.dbs[dbnum];
+		PGconn	   *conn = connectToServer(cluster, dbinfo->db_name);
+		PGresult   *res;
+		AclInfo	   *aclinfos = NULL;
+		AclInfo	   *curr = NULL;
+		int			nacls = 0,
+					size_acls = 8;
+		int			aclnum = 0;
+		int			i_obj_type,
+					i_obj_ident,
+					i_rolname,
+					i_is_relation;
+
+		res = executeQueryOrDie(conn,
+			/*
+			 * Get relations, attributes, functions and procedures.  Some system
+			 * objects like views are not pinned, but these type of objects are
+			 * created in pg_catalog schema.
+			 */
+			"SELECT obj.type, obj.identity, shd.refobjid::regrole rolename, "
+			"    CASE WHEN shd.classid = 'pg_class'::regclass THEN true "
+			"        ELSE false "
+			"    END is_relation "
+			"FROM pg_catalog.pg_shdepend shd, "
+			"LATERAL pg_catalog.pg_identify_object("
+			"    shd.classid, shd.objid, shd.objsubid) obj "
+			/* 'a' is for SHARED_DEPENDENCY_ACL */
+			"WHERE shd.deptype = 'a' AND shd.dbid = %d "
+			"    AND shd.classid IN ('pg_proc'::regclass, 'pg_class'::regclass) "
+			/* get only system objects */
+			"    AND obj.schema = 'pg_catalog' "
+			/*
+			 * Sort only by identity.  It should be enough to uniquely compare
+			 * objects later in check_for_changed_signatures().
+			 * Don't rely on database collation, since we use strcmp
+			 * comparison below.
+			 */
+			"ORDER BY obj.identity COLLATE \"C\";", dbinfo->db_oid);
+
+		i_obj_type = PQfnumber(res, "type");
+		i_obj_ident = PQfnumber(res, "identity");
+		i_rolname = PQfnumber(res, "rolename");
+		i_is_relation = PQfnumber(res, "is_relation");
+
+		aclinfos = (AclInfo *) pg_malloc(sizeof(AclInfo) * size_acls);
+
+		while (aclnum < PQntuples(res))
+		{
+			PQExpBuffer roles_buf;
+
+			if (nacls == size_acls)
+			{
+				size_acls *= 2;
+				aclinfos = (AclInfo *) pg_realloc(aclinfos,
+												  sizeof(AclInfo) * size_acls);
+			}
+			curr = &aclinfos[nacls++];
+
+			curr->obj_type = pg_strdup(PQgetvalue(res, aclnum, i_obj_type));
+			curr->obj_ident = pg_strdup(PQgetvalue(res, aclnum, i_obj_ident));
+			curr->is_relation = PQgetvalue(res, aclnum, i_is_relation)[0] == 't';
+
+			roles_buf = createPQExpBuffer();
+			initPQExpBuffer(roles_buf);
+
+			/*
+			 * For each object gather string of role names mentioned in ACL
+			 * in a format convenient for further use in REVOKE statement.
+			 */
+			while (aclnum < PQntuples(res) &&
+				   strcmp(curr->obj_ident, PQgetvalue(res, aclnum, i_obj_ident)) == 0)
+			{
+				if (roles_buf->len != 0)
+					appendPQExpBufferChar(roles_buf, ',');
+
+				appendPQExpBufferStr(roles_buf,
+					quote_identifier(PQgetvalue(res, aclnum, i_rolname)));
+				aclnum++;
+			}
+
+			curr->role_names = pg_strdup(roles_buf->data);
+			destroyPQExpBuffer(roles_buf);
+		}
+
+		PQclear(res);
+		PQfinish(conn);
+
+		dbinfo->non_def_acl_arr.aclinfos = aclinfos;
+		dbinfo->non_def_acl_arr.nacls = nacls;
+
+		pg_log(PG_REPORT, "get_non_default_acl_infos nacls %d\n", dbinfo->non_def_acl_arr.nacls);
+
+	}
+}
+
+
 /*
  * get_db_infos()
  *
@@ -384,6 +497,10 @@ get_db_infos(ClusterInfo *cluster)
 		dbinfos[tupnum].db_ctype = pg_strdup(PQgetvalue(res, tupnum, i_datctype));
 		snprintf(dbinfos[tupnum].db_tablespace, sizeof(dbinfos[tupnum].db_tablespace), "%s",
 				 PQgetvalue(res, tupnum, i_spclocation));
+
+		/* initialize clean array */
+		dbinfos[tupnum].non_def_acl_arr.nacls = 0;
+		dbinfos[tupnum].non_def_acl_arr.aclinfos = NULL;
 	}
 	PQclear(res);
 
@@ -595,6 +712,9 @@ free_db_and_rel_infos(DbInfoArr *db_arr)
 	for (dbnum = 0; dbnum < db_arr->ndbs; dbnum++)
 	{
 		free_rel_infos(&db_arr->dbs[dbnum].rel_arr);
+
+		if (&db_arr->dbs[dbnum].non_def_acl_arr.nacls > 0)
+			free_acl_infos(&db_arr->dbs[dbnum].non_def_acl_arr);
 		pg_free(db_arr->dbs[dbnum].db_name);
 	}
 	pg_free(db_arr->dbs);
@@ -620,6 +740,25 @@ free_rel_infos(RelInfoArr *rel_arr)
 	rel_arr->nrels = 0;
 }
 
+static void
+free_acl_infos(AclInfoArr *acl_arr)
+{
+	int			aclnum;
+
+	pg_log(PG_REPORT, "free_acl_infos 1 %d\n", acl_arr->nacls);
+	for (aclnum = 0; aclnum < acl_arr->nacls; aclnum++)
+	{
+		pg_free(acl_arr->aclinfos[aclnum].obj_type);
+		pg_free(acl_arr->aclinfos[aclnum].obj_ident);
+		pg_free(acl_arr->aclinfos[aclnum].role_names);
+	}
+
+	pg_free(acl_arr->aclinfos);
+	acl_arr->aclinfos = NULL;
+	acl_arr->nacls = 0;
+	pg_log(PG_REPORT, "free_acl_infos 2 %d\n", acl_arr->nacls);
+}
+
 
 static void
 print_db_infos(DbInfoArr *db_arr)

src/bin/pg_upgrade/pg_upgrade.h

@@ -147,6 +147,26 @@ typedef struct
 	int			nrels;
 } RelInfoArr;
 
+/*
+ * Information about database object needed to check
+ * if its signature has changed between versions
+ * and generate REVOKE statement if necessary.
+ */
+typedef struct
+{
+	char	   *obj_type;		/* object type */
+	char	   *obj_ident;		/* complete object identity */
+	bool		is_relation;	/* if the object is relation */
+	char	   *role_names;		/* list of role names which have permissions
+								 * on the object */
+} AclInfo;
+
+typedef struct
+{
+	AclInfo	   *aclinfos;
+	int			nacls;
+} AclInfoArr;
+
 /*
  * The following structure represents a relation mapping.
  */
@@ -183,6 +203,8 @@ typedef struct
 	char	   *db_ctype;
 	int			db_encoding;
 	RelInfoArr	rel_arr;		/* array of all user relinfos */
+	AclInfoArr	non_def_acl_arr;	/* array of objects info with non default
+									 * ACL */
 } DbInfo;
 
 typedef struct
@@ -390,6 +412,7 @@ FileNameMap *gen_db_file_maps(DbInfo *old_db,
 							  DbInfo *new_db, int *nmaps, const char *old_pgdata,
 							  const char *new_pgdata);
 void		get_db_and_rel_infos(ClusterInfo *cluster);
+void		get_non_default_acl_infos(ClusterInfo *cluster);
 void		print_maps(FileNameMap *maps, int n,
 					   const char *db_name);